t/RN-Praxis3
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Made some obsolete thread creation cleanup

Browse Source
This commit is contained in:
Ano-sys
2025-02-07 18:49:31 +01:00
parent 097ca6a6d3
commit 8774699f4e
6 changed files with 11 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
praxis3/zmq_distributor.c
+9 -81
View File
@@ -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);
}