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

Tried to fix but made things worse

Browse Source
This commit is contained in:
Ano-sys
2025-01-29 23:22:57 +01:00
parent bc1fde1b9f
commit 4910b0db61
26 changed files with 8145 additions and 623 deletions
Download Patch File Download Diff File Expand all files Collapse all files
praxis3/zmq_distributor.c
+134 -27
View File
@@ -21,6 +21,27 @@ char *words_list;
char *get_file_content(FILE *fp)
{
if (!fp) {
perror("Failed to open file");
return NULL;
}
fseek(fp, 0, SEEK_END);
long file_size = ftell(fp);
rewind(fp);
char *content = malloc(file_size + 1);
if (!content) {
perror("Failed to allocate memory for entire file");
fclose(fp);
return NULL;
}
size_t bytes_read = fread(content, 1, file_size, fp);
content[bytes_read] = '\0';
fclose(fp);
return content;
/*
char *buffer = (char*)malloc(sizeof(char) * MESSAGE_LENGTH);
if (buffer == NULL)
{
@@ -57,6 +78,7 @@ char *get_file_content(FILE *fp)
}
return buffer;
*/
}
void *receiverFunc(void *args)
@@ -175,6 +197,7 @@ void *receiverFunc(void *args)
}
words_list = tmp;
strcat(words_list, reply);
}
free(params);
@@ -187,49 +210,121 @@ void *receiverFunc(void *args)
return NULL;
}
void map(size_t worker_count, char **worker_ports, char *filename) {
FILE *fp = fopen(filename, "r");
if (!fp) {
perror("[MAIN - map] Failed to open file");
char *sanitize_input(char *input){
char *ret = malloc(sizeof(char));
if(!ret){
perror("[MAIN - sanitize_input] Failed to allocate memory for ret");
exit(1);
}
*ret = '\0';
fseek(fp, 0, SEEK_END);
size_t file_size = ftell(fp);
rewind(fp);
char *it = input;
while(!isalpha(*it)) it++;
size_t worker_index = 0;
size_t thread_count = 0;
int i = 0;
bool writeSpace = false;
while(*it){
if(isalpha(*it)){
char *tmp = NULL;
if(writeSpace){
tmp = realloc(ret, strlen(ret) + 2);
}
else{
tmp = realloc(ret, strlen(ret) + 1);
}
if(!tmp){
perror("[MAIN - sanitize_input] Failed to realloc memory for ret");
free(ret);
exit(1);
}
ret = tmp;
if(writeSpace) ret[i++] = ' ';
ret[i++] = tolower(*it++);
writeSpace = false;
}
else{
it++;
writeSpace = true;
}
}
ret[strlen(ret) + 1] = '\0';
return ret;
}
pthread_t *threads = (pthread_t*)malloc(sizeof(pthread_t) * worker_count * 2);
if (!threads) {
perror("[MAIN - map] Failed to allocate memory for threads");
fclose(fp);
void map(size_t worker_count, char **worker_ports, const char *filename) {
FILE *fp = fopen(filename, "rb");
char *raw = get_file_content(fp);
if(!raw){
perror("Failed to read file");
exit(1);
}
char *content = sanitize_input(raw);
free(raw);
if (!content) {
pthread_mutex_destroy(&words_mutex);
exit(1);
}
char buffer[MESSAGE_LENGTH + 1];
size_t max_threads = worker_count * 2;
pthread_t *threads = malloc(max_threads * sizeof(pthread_t));
if (!threads) {
perror("[MAIN - map] Failed to allocate memory for threads");
free(content);
pthread_mutex_destroy(&words_mutex);
exit(1);
}
while (!feof(fp)) {
size_t bytes_read = fread(buffer, sizeof(char), MESSAGE_LENGTH, fp);
if (bytes_read == 0) break;
size_t thread_count = 0;
size_t worker_index = 0;
buffer[bytes_read] = '\0';
size_t total_len = strlen(content);
size_t offset = 0;
THREAD_ARGS *args = (THREAD_ARGS*)malloc(sizeof(THREAD_ARGS));
while (offset < total_len) {
size_t chunk_size = (total_len - offset >= MESSAGE_LENGTH) ? MESSAGE_LENGTH : (total_len - offset);
if (offset + chunk_size < total_len) {
size_t new_chunk_size = chunk_size;
while (new_chunk_size > 0) {
char c = content[offset + new_chunk_size];
if (!isalpha((unsigned char)c)) {
break;
}
new_chunk_size--;
}
if (new_chunk_size == 0) {
new_chunk_size = chunk_size;
}
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);
fclose(fp);
free(content);
pthread_mutex_destroy(&words_mutex);
exit(1);
}
args->port = atol(worker_ports[worker_index]);
args->data = strdup(buffer);
args->type = MAP;
args->data = strndup(content + offset, chunk_size);
if (pthread_create(&threads[thread_count], NULL, receiverFunc, (void*)args) != 0) {
perror("[MAIN - map] Failed to create thread");
free(args->data);
@@ -237,20 +332,19 @@ void map(size_t worker_count, char **worker_ports, char *filename) {
continue;
}
offset += chunk_size;
worker_index = (worker_index + 1) % worker_count;
thread_count++;
}
fclose(fp);
for (size_t i = 0; i < thread_count; i++) {
pthread_join(threads[i], NULL);
}
free(threads);
free(content);
}
char *extract_reduce_string(const char * toExtract){
if(!toExtract) return NULL;
size_t toExtract_len = strlen(toExtract);
@@ -269,7 +363,6 @@ char *extract_reduce_string(const char * toExtract){
return extracted;
}
void reduce(char *toReduce, size_t worker_count, char **worker_ports) {
if (!toReduce || worker_count == 0) {
fprintf(stderr, "[MAIN - reduce] Invalid input parameters.\n");
@@ -289,7 +382,22 @@ void reduce(char *toReduce, size_t worker_count, char **worker_ports) {
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 chunk_size = (toReduce_len - offset >= MESSAGE_LENGTH) ? MESSAGE_LENGTH : (toReduce_len - offset);
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)) {
break;
}
new_size--;
}
if (new_size == 0) {
new_size = chunk_size;
}
chunk_size = new_size;
}
THREAD_ARGS *args = (THREAD_ARGS*)malloc(sizeof(THREAD_ARGS));
if (!args) {
@@ -454,7 +562,6 @@ int main(const int argc, char **argv) {
rip(argc - 2, argv + 2);
// print words_list
// TODO: sort by frequency and alphabetically
print_words_list(word_list_reduced);
free(word_list_reduced);