diff --git a/praxis3/build/CMakeFiles/zmq_distributor.dir/zmq_distributor.c.o b/praxis3/build/CMakeFiles/zmq_distributor.dir/zmq_distributor.c.o index ae7d6cd..83cd38f 100644 Binary files a/praxis3/build/CMakeFiles/zmq_distributor.dir/zmq_distributor.c.o and b/praxis3/build/CMakeFiles/zmq_distributor.dir/zmq_distributor.c.o differ diff --git a/praxis3/build/zmq_distributor b/praxis3/build/zmq_distributor index b265472..550604f 100755 Binary files a/praxis3/build/zmq_distributor and b/praxis3/build/zmq_distributor differ diff --git a/praxis3/cmake-build-debug/CMakeFiles/zmq_distributor.dir/zmq_distributor.c.o b/praxis3/cmake-build-debug/CMakeFiles/zmq_distributor.dir/zmq_distributor.c.o index db46cab..17653a5 100644 Binary files a/praxis3/cmake-build-debug/CMakeFiles/zmq_distributor.dir/zmq_distributor.c.o and b/praxis3/cmake-build-debug/CMakeFiles/zmq_distributor.dir/zmq_distributor.c.o differ diff --git a/praxis3/cmake-build-debug/Testing/Temporary/LastTest.log b/praxis3/cmake-build-debug/Testing/Temporary/LastTest.log index 4b5286f..6d2a0c2 100644 --- a/praxis3/cmake-build-debug/Testing/Temporary/LastTest.log +++ b/praxis3/cmake-build-debug/Testing/Temporary/LastTest.log @@ -1,3 +1,3 @@ -Start testing: Feb 07 18:31 CET +Start testing: Feb 07 18:41 CET ---------------------------------------------------------- -End testing: Feb 07 18:31 CET +End testing: Feb 07 18:41 CET diff --git a/praxis3/cmake-build-debug/zmq_distributor b/praxis3/cmake-build-debug/zmq_distributor index 16e9f76..d9333aa 100755 Binary files a/praxis3/cmake-build-debug/zmq_distributor and b/praxis3/cmake-build-debug/zmq_distributor differ diff --git a/praxis3/zmq_distributor.c b/praxis3/zmq_distributor.c index 5c8d406..9a5e2af 100644 --- a/praxis3/zmq_distributor.c +++ b/praxis3/zmq_distributor.c @@ -278,20 +278,8 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c // perror("Failed to read file"); exit(1); } - /* - if(strcmp(raw, "") == 0 || strcmp(raw, "\n") == 0){ - printf("word,frequency"); - exit(0); - } - */ - char *content = raw; // sanitize_input(raw); - // free(raw); - if (!content) { - pthread_mutex_destroy(&words_mutex); - pthread_mutex_destroy(&context_mutex); - pthread_mutex_destroy(&thread_count_mutex); - exit(1); - } + + char *content = raw; if(strcmp(content, "") == 0){ fucking_string_empty(context, worker_ports); @@ -299,18 +287,7 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c return; } - size_t max_threads = worker_count * 2; - pthread_t *threads = malloc(sizeof(pthread_t) * max_threads); - if (!threads) { - // perror("[MAIN - map] Failed to allocate memory for threads"); - free(content); - pthread_mutex_destroy(&words_mutex); - exit(1); - } - - size_t thread_count = 0; size_t worker_index = 0; - size_t total_len = strlen(content); size_t offset = 0; @@ -332,23 +309,9 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c chunk_size = new_chunk_size; } - if (thread_count == max_threads) { - max_threads *= 2; - pthread_t *tmp = realloc(threads, max_threads * sizeof(pthread_t)); - if (!tmp) { - // perror("[MAIN - map] Failed to reallocate memory for threads"); - free(threads); - free(content); - pthread_mutex_destroy(&words_mutex); - exit(1); - } - threads = tmp; - } - THREAD_ARGS *args = malloc(sizeof(THREAD_ARGS)); if (!args) { // perror("[MAIN - map] Failed to allocate memory for args"); - free(threads); free(content); pthread_mutex_destroy(&words_mutex); exit(1); @@ -370,27 +333,21 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c pthread_mutex_unlock(&thread_count_mutex); } - if (pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0) { + pthread_t thread; + if (pthread_create(&thread, NULL, receiverFunc, (void*)args) != 0) { // perror("[MAIN - map] Failed to create thread"); free(args->data); free(args); continue; } + pthread_detach(thread); offset += chunk_size; worker_index = (worker_index + 1) % worker_count; - thread_count++; - alterRuntimeThreadCount(1); } free(content); - /* - for (size_t i = 0; i < thread_count; i++) { - pthread_join(threads[i], NULL); - } - */ - pthread_mutex_lock(&thread_count_mutex); int unlocked_runtime_thread_count = RUNTIME_THREAD_COUNT; pthread_mutex_unlock(&thread_count_mutex); @@ -400,8 +357,6 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c unlocked_runtime_thread_count = RUNTIME_THREAD_COUNT; pthread_mutex_unlock(&thread_count_mutex); } - - free(threads); } char *extract_reduce_string(const char * toExtract){ @@ -428,18 +383,9 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void return; } - size_t max_threads = worker_count * 2; - pthread_t *threads = malloc(sizeof(pthread_t) * max_threads); - if (!threads) { - // perror("[MAIN - reduce] Failed to allocate memory for threads"); - pthread_mutex_destroy(&words_mutex); - exit(1); - } - size_t toReduce_len = strlen(toReduce); size_t offset = 0; size_t worker_index = 0; - size_t thread_count = 0; while (offset < toReduce_len) { size_t remaining = toReduce_len - offset; @@ -464,7 +410,6 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void 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); } @@ -490,37 +435,22 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void pthread_mutex_unlock(&thread_count_mutex); } - if (pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0) { + pthread_t thread; + if (pthread_create(&thread, NULL, receiverFunc, (void*)args) != 0) { // perror("[MAIN - reduce] Failed to create thread"); free(args->data); free(args); continue; } + pthread_detach(thread); + offset += chunk_size; worker_index = (worker_index + 1) % worker_count; - thread_count++; alterRuntimeThreadCount(1); - - if (thread_count == max_threads){ - max_threads *= 2; - pthread_t *tmp = realloc(threads, sizeof(pthread_t) * max_threads); - if (!tmp) { - // perror("[MAIN - map] Failed to reallocate memory for threads"); - free(threads); - pthread_mutex_destroy(&words_mutex); - exit(1); - } - threads = tmp; - } } - /* - for (size_t i = 0; i < thread_count; i++) { - pthread_join(threads[i], NULL); - } - */ pthread_mutex_lock(&thread_count_mutex); int unlocked_runtime_thread_count = RUNTIME_THREAD_COUNT; pthread_mutex_unlock(&thread_count_mutex); @@ -530,8 +460,6 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void unlocked_runtime_thread_count = RUNTIME_THREAD_COUNT; pthread_mutex_unlock(&thread_count_mutex); } - - free(threads); }