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

No memory leaks anymore

Browse Source
This commit is contained in:
Ano-sys
2025-02-02 16:51:59 +01:00
parent 7f6d6c1479
commit 02c31d009e
60 changed files with 474261 additions and 57129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
praxis3/zmq_distributor.c
+144 -89
View File
@@ -17,6 +17,7 @@ typedef struct{
}THREAD_ARGS;
pthread_mutex_t words_mutex;
pthread_mutex_t context_mutex;
char *words_list;
@@ -25,7 +26,7 @@ char *sanitize_input(char *input);
char *get_file_content(FILE *fp)
{
if (!fp) {
perror("Failed to open file");
// perror("Failed to open file");
return NULL;
}
fseek(fp, 0, SEEK_END);
@@ -34,7 +35,7 @@ char *get_file_content(FILE *fp)
char *content = malloc(file_size + 1);
if (!content) {
perror("Failed to allocate memory for entire file");
// perror("Failed to allocate memory for entire file");
fclose(fp);
return NULL;
}
@@ -48,14 +49,14 @@ char *get_file_content(FILE *fp)
char *buffer = (char*)malloc(sizeof(char) * MESSAGE_LENGTH);
if (buffer == NULL)
{
perror("Failed to allocate memory for buffer");
// perror("Failed to allocate memory for buffer");
return NULL;
}
const size_t bytes_read = fread(buffer, sizeof(char), MESSAGE_LENGTH - 1, fp);
if (bytes_read == 0)
{
perror("File is empty or nothing is left to read");
// perror("File is empty or nothing is left to read");
free(buffer);
return NULL;
}
@@ -67,7 +68,7 @@ char *get_file_content(FILE *fp)
if (delimiter_pos < 0)
{
fprintf(stderr, "No delimiting character was found -> taking whole buffer");
// fprintf(stderr, "No delimiting character was found -> taking whole buffer");
delimiter_pos = (int)bytes_read;
}
@@ -75,7 +76,7 @@ char *get_file_content(FILE *fp)
if (fseek(fp, delimiter_pos - (int) bytes_read + 2, SEEK_CUR) != 0)
{
perror("Failed to seek in file");
// perror("Failed to seek in file");
free(buffer);
return NULL;
}
@@ -88,12 +89,13 @@ void *receiverFunc(void *args)
{
THREAD_ARGS *params = (THREAD_ARGS*)args;
if(!params){
perror("[THREAD] invalid arguments");
// perror("[THREAD] invalid arguments");
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);
return NULL;
}
@@ -101,53 +103,66 @@ void *receiverFunc(void *args)
void *context = params->context;
if(!context){
perror("[THREAD] Failed to create context");
// perror("[THREAD] Failed to create context");
free(params->data);
free(params);
return NULL;
}
pthread_mutex_lock(&context_mutex);
void *requester = zmq_socket(context, ZMQ_REQ);
if(!requester){
perror("[THREAD] Failed to create socket");
// perror("[THREAD] Failed to create socket");
free(params->data);
free(params);
pthread_mutex_unlock(&context_mutex);
zmq_ctx_destroy(context);
return NULL;
}
pthread_mutex_unlock(&context_mutex);
char addr[22];
snprintf(addr, 21, "tcp://localhost:%d", params->port); // BUG: FIX PORT 0
pthread_mutex_lock(&context_mutex);
int rc = zmq_connect(requester, addr);
if (rc != 0){
perror("Failed to connect");
free(params);
zmq_close(requester);
zmq_ctx_destroy(context);
return NULL;
}
fprintf(stderr, "[THREAD] Connected to worker: %s\n", addr);
char *data = sanitize_input(params->data);
if(!data){
perror("[THREAD] Failed to sanitize input");
// perror("Failed to connect");
free(params->data);
free(params);
zmq_close(requester);
pthread_mutex_unlock(&context_mutex);
return NULL;
}
pthread_mutex_unlock(&context_mutex);
// fprintf(stderr, "[THREAD] Connected to worker: %s\n", addr);
char *data = NULL;
if(params->type == MAP) data = sanitize_input(params->data);
else data = strdup(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);
char request[len + 1];
/*
if(!request){
perror("[THREAD] Failed to allocate memory for request");
// perror("[THREAD] Failed to allocate memory for request");
free(params);
free(data);
zmq_close(requester);
zmq_ctx_destroy(context);
return NULL;
}
*/
switch(params->type){
@@ -161,47 +176,46 @@ 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);
// 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");
// fprintf(stderr, "[THREAD] Send request: %s\n", request);
pthread_mutex_lock(&context_mutex);
if(zmq_send(requester, request, strlen(request) + 1, 0) == -1){
// perror("[THREAD] Failed to send request");
free(params);
free(request);
// free(request);
zmq_close(requester);
zmq_ctx_destroy(context);
pthread_mutex_unlock(&context_mutex);
return NULL;
}
pthread_mutex_unlock(&context_mutex);
free(request);
// free(request);
fprintf(stderr, "[THREAD]: Waiting for response\n");
// fprintf(stderr, "[THREAD]: Waiting for response\n");
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");
// perror("[THREAD] Failed to receive reply");
free(params);
zmq_close(requester);
zmq_ctx_destroy(context);
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);
zmq_ctx_destroy(context);
return NULL;
}
@@ -211,22 +225,20 @@ 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");
free(params);
pthread_mutex_unlock(&words_mutex);
zmq_close(requester);
zmq_ctx_destroy(context);
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");
free(params);
pthread_mutex_unlock(&words_mutex);
zmq_close(requester);
zmq_ctx_destroy(context);
return NULL;
}
@@ -250,13 +262,13 @@ char *sanitize_input(char *input){
char *ret = malloc(sizeof(char));
if(!ret){
perror("[MAIN - sanitize_input] Failed to allocate memory for ret");
exit(1);
// perror("[MAIN - sanitize_input] Failed to allocate memory for ret");
return NULL;
}
*ret = '\0';
char *it = input;
while(!isalpha(*it)) it++;
while(*it && !isalpha(*it)) it++;
bool writeSpace = false;
while(*it){
@@ -269,7 +281,7 @@ char *sanitize_input(char *input){
tmp = realloc(ret, i + 2);
}
if(!tmp){
perror("[MAIN - sanitize_input] Failed to realloc memory for ret");
// perror("[MAIN - sanitize_input] Failed to realloc memory for ret");
free(ret);
return NULL;
}
@@ -291,20 +303,27 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c
FILE *fp = fopen(filename, "rb");
char *raw = get_file_content(fp);
if(!raw){
perror("Failed to read file");
// perror("Failed to read file");
exit(1);
}
/*
if(strcmp(raw, "") == 0 || strcmp(raw, "\n") == 0){
printf("word,frequency");
exit(0);
}
*/
char *content = sanitize_input(raw);
free(raw);
if (!content) {
pthread_mutex_destroy(&words_mutex);
pthread_mutex_destroy(&context_mutex);
exit(1);
}
size_t max_threads = worker_count * 2;
pthread_t *threads = malloc(max_threads * sizeof(pthread_t));
pthread_t *threads = malloc(sizeof(pthread_t) * max_threads);
if (!threads) {
perror("[MAIN - map] Failed to allocate memory for threads");
// perror("[MAIN - map] Failed to allocate memory for threads");
free(content);
pthread_mutex_destroy(&words_mutex);
exit(1);
@@ -316,8 +335,8 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c
size_t total_len = strlen(content);
size_t offset = 0;
while (offset < total_len) {
size_t chunk_size = (total_len - offset >= PAYLOAD_LENGTH + 1) ? PAYLOAD_LENGTH + 1 : (total_len - offset);
while (offset < total_len || (offset == 0 && total_len == 0)) {
size_t chunk_size = (total_len - offset >= PAYLOAD_LENGTH) ? PAYLOAD_LENGTH : (total_len - offset);
if (offset + chunk_size < total_len) {
size_t new_chunk_size = chunk_size;
@@ -338,7 +357,7 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c
max_threads *= 2;
pthread_t *tmp = realloc(threads, max_threads * sizeof(pthread_t));
if (!tmp) {
perror("[MAIN - map] Failed to reallocate memory for threads");
// perror("[MAIN - map] Failed to reallocate memory for threads");
free(threads);
free(content);
pthread_mutex_destroy(&words_mutex);
@@ -349,7 +368,7 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c
THREAD_ARGS *args = malloc(sizeof(THREAD_ARGS));
if (!args) {
perror("[MAIN - map] Failed to allocate memory for args");
// perror("[MAIN - map] Failed to allocate memory for args");
free(threads);
free(content);
pthread_mutex_destroy(&words_mutex);
@@ -363,7 +382,7 @@ void map(size_t worker_count, char **worker_ports, const char *filename, void *c
args->context = context;
if (pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0) {
perror("[MAIN - map] Failed to create thread");
// perror("[MAIN - map] Failed to create thread");
free(args->data);
free(args);
continue;
@@ -394,7 +413,7 @@ char *extract_reduce_string(const char * toExtract){
char *extracted = strndup(toExtract, copy_length);
if(!extracted){
perror("[MAIN - reduce - extract-reduce-string] Failed to allocate memory for extracted string");
// perror("[MAIN - reduce - extract-reduce-string] Failed to allocate memory for extracted string");
return NULL;
}
return extracted;
@@ -402,13 +421,14 @@ char *extract_reduce_string(const char * toExtract){
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");
// fprintf(stderr, "[MAIN - reduce] Invalid input parameters.\n");
return;
}
pthread_t *threads = malloc(sizeof(pthread_t) * worker_count * 2);
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");
// perror("[MAIN - reduce] Failed to allocate memory for threads");
pthread_mutex_destroy(&words_mutex);
exit(1);
}
@@ -419,27 +439,24 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void
size_t thread_count = 0;
while (offset < toReduce_len) {
size_t chunk_size = (toReduce_len - offset >= MESSAGE_LENGTH) ? MESSAGE_LENGTH : (toReduce_len - offset);
size_t remaining = toReduce_len - offset;
size_t chunk_size = (remaining >= PAYLOAD_LENGTH) ? PAYLOAD_LENGTH : remaining;
if (offset + chunk_size < toReduce_len) {
size_t new_size = chunk_size;
while (new_size > 0) {
char c = toReduce[offset + new_size];
if (isdigit((unsigned char)c)) {
new_size++;
break;
}
new_size--;
size_t adjust = chunk_size;
while (adjust > 0 && !isdigit((unsigned char)toReduce[offset + adjust - 1])) {
adjust--;
}
if (new_size == 0) {
new_size = chunk_size;
if (adjust > 0) {
chunk_size = adjust;
}
chunk_size = new_size;
}
THREAD_ARGS *args = (THREAD_ARGS*)malloc(sizeof(THREAD_ARGS));
if (!args) {
perror("[MAIN - reduce] Failed to allocate memory for args");
// perror("[MAIN - reduce] Failed to allocate memory for args");
free(threads);
pthread_mutex_destroy(&words_mutex);
exit(1);
@@ -451,13 +468,13 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void
args->context = context;
if (!args->data) {
perror("[MAIN - reduce] Failed to allocate memory for chunk data");
// perror("[MAIN - reduce] Failed to allocate memory for chunk data");
free(args);
continue;
}
if (pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0) {
perror("[MAIN - reduce] Failed to create thread");
// perror("[MAIN - reduce] Failed to create thread");
free(args->data);
free(args);
continue;
@@ -466,6 +483,18 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void
offset += chunk_size;
worker_index = (worker_index + 1) % worker_count;
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);
pthread_mutex_destroy(&words_mutex);
exit(1);
}
threads = tmp;
}
}
for (size_t i = 0; i < thread_count; i++) {
@@ -477,46 +506,61 @@ void reduce(const char *toReduce, size_t worker_count, char **worker_ports, void
void rip(size_t worker_count, char **worker_ports, void *context){
pthread_t *threads = malloc(sizeof(pthread_t) * worker_count);
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");
// perror("[MAIN- reduce] Failed to allocate memory for threads");
pthread_mutex_destroy(&words_mutex);
exit(1);
}
size_t thread_count = 0;
for(int i = 0; i < worker_count; i++){
for(size_t i = 0; i < worker_count; i++){
THREAD_ARGS *args = (THREAD_ARGS*)malloc(sizeof(THREAD_ARGS));
if(!args){
perror("[MAIN - rip] Failed to allocate memory for args");
// perror("[MAIN - rip] Failed to allocate memory for args");
free(threads);
pthread_mutex_destroy(&words_mutex);
exit(1);
}
args->port = atol(worker_ports[i]);
args->port = atol(worker_ports[(int)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");
// fprintf(stderr, "[MAIN - reduce] Warning: extract_reduce_string() returned NULL\n");
free(args->data);
free(args);
continue;
}
if(pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0){
perror("[MAIN - reduce] Failed to create thread");
// perror("[MAIN - reduce] Failed to create thread");
free(args->data);
free(args);
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);
pthread_mutex_destroy(&words_mutex);
exit(1);
}
threads = tmp;
}
}
// wait for workers to finish
for(int i = 0; i < thread_count; i++){
pthread_join(threads[i], NULL);
for(size_t i = 0; i < thread_count; i++){
pthread_join(threads[(int)i], NULL);
}
free(threads);
@@ -538,7 +582,8 @@ int qsort_sort_comparer(const void *kv1, const void *kv2){
void print_words_list(char *list){
key_value_pair *head = NULL, *tail = NULL;
extract_key_value_pairs(list, &head, &tail, false);
extract_key_value_pairs(list, &head, &tail, false); // TODO: afterwords recount values ... und32 -> und5
// defragment_key_value_pairs(&head, &tail);
size_t count = 0;
for (key_value_pair *it = head; it != NULL; it = it->next) {
@@ -552,7 +597,7 @@ void print_words_list(char *list){
key_value_pair **kv_array = malloc(count * sizeof(key_value_pair *));
if (!kv_array) {
perror("[MAIN - print_words_list] Failed to allocate memory for sorting");
// perror("[MAIN - print_words_list] Failed to allocate memory for sorting");
return;
}
@@ -574,22 +619,25 @@ void print_words_list(char *list){
int main(const int argc, char **argv) {
if(argc < 3){
fprintf(stderr, "Usage: %s <file.txt> <worker port 1> <worker port 2> ... <worker port n>", argv[0]);
// fprintf(stderr, "Usage: %s <file.txt> <worker port 1> <worker port 2> ... <worker port n>", argv[0]);
exit(1);
}
pthread_mutex_init(&words_mutex, NULL);
pthread_mutex_init(&context_mutex, NULL);
void *context = zmq_ctx_new();
if(!context){
perror("[MAIN] zmq_ctx_new failed");
// perror("[MAIN] zmq_ctx_new failed");
exit(1);
}
// MAP
map(argc - 2, argv + 2, argv[1], context);
fprintf(stderr, "MAP:\n%s\n", words_list);
// fprintf(stderr, "MAP:\n%s\n", words_list);
if(words_list == NULL){
words_list = strdup("");
}
char *word_list_mapped = strdup(words_list);
free(words_list);
words_list = NULL;
@@ -597,12 +645,17 @@ int main(const int argc, char **argv) {
// REDUCE
reduce(word_list_mapped, argc - 2, argv + 2, context);
free(word_list_mapped);
word_list_mapped = NULL;
if(words_list == NULL){
words_list = strdup("");
}
char *word_list_reduced = strdup(words_list);
free(words_list);
words_list = NULL;
fprintf(stderr, "REDUCE:\n%s\n", word_list_reduced);
// fprintf(stderr, "REDUCE:\n%s\n", word_list_reduced);
// RIP
rip(argc - 2, argv + 2, context);
@@ -610,8 +663,10 @@ int main(const int argc, char **argv) {
// print words_list
print_words_list(word_list_reduced);
free(word_list_reduced);
word_list_reduced = NULL;
pthread_mutex_destroy(&words_mutex);
pthread_mutex_destroy(&context_mutex);
zmq_ctx_destroy(context);
return 0;