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

Seems better by now -> TODO cleanup

Browse Source
This commit is contained in:
Ano-sys
2025-02-07 18:33:46 +01:00
parent f856a4fa82
commit 097ca6a6d3
208 changed files with 135393 additions and 639928 deletions
Download Patch File Download Diff File Expand all files Collapse all files
praxis3/zmq_distributor.c
+89 -25
View File
@@ -18,8 +18,11 @@ typedef struct{
pthread_mutex_t words_mutex;
pthread_mutex_t context_mutex;
pthread_mutex_t thread_count_mutex;
char *words_list;
int RUNTIME_THREAD_COUNT = 0;
int MAX_THREAD_LIMIT = 0;
char *get_file_content(FILE *fp)
{
@@ -38,8 +41,8 @@ char *get_file_content(FILE *fp)
return NULL;
}
size_t bytes_read = fread(content, 1, file_size, fp);
content[bytes_read] = '\0';
fread(content, file_size, 1, fp);
content[file_size] = '\0';
fclose(fp);
return content;
@@ -83,36 +86,46 @@ char *get_file_content(FILE *fp)
*/
}
void alterRuntimeThreadCount(int yoda){
pthread_mutex_lock(&thread_count_mutex);
RUNTIME_THREAD_COUNT += yoda;
pthread_mutex_unlock(&thread_count_mutex);
}
void *receiverFunc(void *args)
{
THREAD_ARGS *params = (THREAD_ARGS*)args;
if(!params){
// perror("[THREAD] invalid arguments");
perror("[THREAD] invalid arguments");
alterRuntimeThreadCount(-1);
return NULL;
}
if (params->port <= 0) {
// fprintf(stderr, "[THREAD] Invalid port: %d\n", params->port);
fprintf(stderr, "[THREAD] Invalid port: %d\n", params->port);
free(params->data);
free(params);
alterRuntimeThreadCount(-1);
return NULL;
}
void *context = params->context;
if(!context){
// perror("[THREAD] Failed to create context");
perror("[THREAD] Failed to create context");
free(params->data);
free(params);
alterRuntimeThreadCount(-1);
return NULL;
}
pthread_mutex_lock(&context_mutex);
void *requester = zmq_socket(context, ZMQ_REQ);
pthread_mutex_unlock(&context_mutex);
if(!requester){
// perror("[THREAD] Failed to create socket");
perror("[THREAD] Failed to create socket");
free(params->data);
free(params);
alterRuntimeThreadCount(-1);
return NULL;
}
@@ -129,6 +142,7 @@ void *receiverFunc(void *args)
free(params->data);
free(params);
zmq_close(requester);
alterRuntimeThreadCount(-1);
return NULL;
}
// fprintf(stderr, "[THREAD] Connected to worker: %s\n", addr);
@@ -140,9 +154,10 @@ void *receiverFunc(void *args)
free(params->data);
if(!data){
// perror("[THREAD] Failed to sanitize input");
perror("[THREAD] Failed to dup input");
free(params);
zmq_close(requester);
alterRuntimeThreadCount(-1);
return NULL;
}
@@ -160,11 +175,12 @@ void *receiverFunc(void *args)
snprintf(request, len, "rip");
break;
default:
// perror("[THREAD] Got unsupported request");
perror("[THREAD] Got unsupported request");
free(data);
free(params);
// free(request);
zmq_close(requester);
alterRuntimeThreadCount(-1);
return NULL;
}
@@ -172,9 +188,10 @@ void *receiverFunc(void *args)
// fprintf(stderr, "[THREAD] Send request: %s\n", request);
if(zmq_send(requester, request, strlen(request) + 1, 0) == -1){
// perror("[THREAD] Failed to send request");
perror("[THREAD] Failed to send request");
free(params);
zmq_close(requester);
alterRuntimeThreadCount(-1);
return NULL;
}
@@ -182,18 +199,20 @@ void *receiverFunc(void *args)
char reply[MESSAGE_LENGTH];
int bytes_recv = zmq_recv(requester, reply, MESSAGE_LENGTH, 0);
if (bytes_recv == -1){
// perror("[THREAD] Failed to receive reply");
perror("[THREAD] Failed to receive reply");
free(params);
zmq_close(requester);
alterRuntimeThreadCount(-1);
return NULL;
}
// reply[bytes_recv] = '\0';
if(strcmp(reply, "rip") == 0){
// perror("[THREAD] Got rip -> shutting down");
perror("[THREAD] Got rip -> shutting down");
free(params);
zmq_close(requester);
alterRuntimeThreadCount(-1);
return NULL;
}
@@ -203,20 +222,22 @@ void *receiverFunc(void *args)
if(!words_list){
words_list = strdup(reply);
if(!words_list){
// perror("[THREAD] Failed to allocate memory for words_list");
perror("[THREAD] Failed to allocate memory for words_list");
pthread_mutex_unlock(&words_mutex);
free(params);
zmq_close(requester);
alterRuntimeThreadCount(-1);
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");
perror("[THREAD] Failed to realloc memory for words_list");
pthread_mutex_unlock(&words_mutex);
free(params);
zmq_close(requester);
alterRuntimeThreadCount(-1);
return NULL;
}
@@ -229,6 +250,8 @@ void *receiverFunc(void *args)
free(params);
zmq_close(requester);
alterRuntimeThreadCount(-1);
return NULL;
}
@@ -266,6 +289,7 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c
if (!content) {
pthread_mutex_destroy(&words_mutex);
pthread_mutex_destroy(&context_mutex);
pthread_mutex_destroy(&thread_count_mutex);
exit(1);
}
@@ -336,6 +360,16 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c
args->data = strndup(content + offset, chunk_size);
args->context = context;
pthread_mutex_lock(&thread_count_mutex);
int current_thread_count = RUNTIME_THREAD_COUNT;
pthread_mutex_unlock(&thread_count_mutex);
while(current_thread_count == MAX_THREAD_LIMIT){
pthread_mutex_lock(&thread_count_mutex);
current_thread_count = RUNTIME_THREAD_COUNT;
pthread_mutex_unlock(&thread_count_mutex);
}
if (pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0) {
// perror("[MAIN - map] Failed to create thread");
free(args->data);
@@ -346,12 +380,26 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c
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);
while(unlocked_runtime_thread_count > 0){
pthread_mutex_lock(&thread_count_mutex);
unlocked_runtime_thread_count = RUNTIME_THREAD_COUNT;
pthread_mutex_unlock(&thread_count_mutex);
}
free(threads);
}
@@ -432,6 +480,16 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void
continue;
}
pthread_mutex_lock(&thread_count_mutex);
int current_thread_count = RUNTIME_THREAD_COUNT;
pthread_mutex_unlock(&thread_count_mutex);
while(current_thread_count == MAX_THREAD_LIMIT){
pthread_mutex_lock(&thread_count_mutex);
current_thread_count = RUNTIME_THREAD_COUNT;
pthread_mutex_unlock(&thread_count_mutex);
}
if (pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0) {
// perror("[MAIN - reduce] Failed to create thread");
free(args->data);
@@ -443,6 +501,8 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void
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);
@@ -456,9 +516,20 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void
}
}
/*
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);
while(unlocked_runtime_thread_count > 0){
pthread_mutex_lock(&thread_count_mutex);
unlocked_runtime_thread_count = RUNTIME_THREAD_COUNT;
pthread_mutex_unlock(&thread_count_mutex);
}
free(threads);
}
@@ -466,7 +537,7 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void
void rip(size_t worker_count, char **worker_ports, void *context){
size_t max_threads = worker_count * 2;
size_t max_threads = worker_count;
pthread_t *threads = malloc(sizeof(pthread_t) * max_threads);
if(!threads){
// perror("[MAIN- reduce] Failed to allocate memory for threads");
@@ -501,17 +572,6 @@ void rip(size_t worker_count, char **worker_ports, void *context){
continue;
}
thread_count++;
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);
return;
}
threads = tmp;
}
}
// wait for workers to finish
@@ -579,8 +639,11 @@ int main(const int argc, char **argv) {
exit(1);
}
MAX_THREAD_LIMIT = argc - 2;
pthread_mutex_init(&words_mutex, NULL);
pthread_mutex_init(&context_mutex, NULL);
pthread_mutex_init(&thread_count_mutex, NULL);
void *context = zmq_ctx_new();
if(!context){
@@ -627,6 +690,7 @@ int main(const int argc, char **argv) {
pthread_mutex_destroy(&words_mutex);
pthread_mutex_destroy(&context_mutex);
pthread_mutex_destroy(&thread_count_mutex);
zmq_ctx_destroy(context);
return 0;