fix: bring project back to a working state

This commit is contained in:
Theo Tappe
2026-07-04 17:55:03 +02:00
parent 75d5f91c6a
commit 4d6171d6bf
17 changed files with 315 additions and 226 deletions
+67 -45
View File
@@ -6,6 +6,8 @@
#include "chunk.h"
#include "config.h"
#include "data.h"
#include "file.h"
#include "log.h"
#include "multiprocessing.h"
#include "queue.h"
@@ -15,13 +17,19 @@
#include <dirent.h>
int send_chunk(Client *client, Chunk *chunk, Config *config) {
if (config->use_compression) {
if (config->use_compression && config->use_chunk_serialization) {
Data *data = chunk_compress(chunk, config->compression_level);
send_data(client->file_descriptor, data->data, data->size);
} else {
for (int i = 0; i < chunk->element_count; i++) {
send_status(client->file_descriptor, NEXT);
send_status(client->file_descriptor, STATUS_NEXT);
File *file = chunk->items[i];
if (config->use_compression) {
Data *compressed_data =
data_compress(file->data, config->compression_level);
data_destroy(file->data);
file->data = compressed_data;
}
file_send_single_calls(file, client->file_descriptor);
}
}
@@ -76,7 +84,11 @@ int load_files_multithreaded(void *pipeline_context) {
int send_chunks_multithreaded(void *pipeline_context) {
PipelineContextSender *context = (PipelineContextSender *)pipeline_context;
Client *client = client_create();
client_connect(client, "127.0.0.1", 8080);
const char *env_ip = getenv("FASTSYNC_SERVER_IP");
const char *ip = env_ip ? env_ip : "127.0.0.1";
const char *env_port = getenv("FASTSYNC_SERVER_PORT");
int port = env_port ? atoi(env_port) : 8080;
client_connect(client, (char *)ip, port);
config_send(client->file_descriptor, context->config);
while (true) {
@@ -85,7 +97,7 @@ int send_chunks_multithreaded(void *pipeline_context) {
&context->condition_not_empty_loader,
&context->condition_not_full_loader, &context->loader_done);
if (current_chunk == NULL) {
send_status(client->file_descriptor, FINISHED);
send_status(client->file_descriptor, STATUS_FINISHED);
client_disconnect(client);
client_delete(client);
return thrd_success;
@@ -100,7 +112,11 @@ int send_chunks_multithreaded(void *pipeline_context) {
int send_files(Config *config) {
Client *client = client_create();
client_connect(client, "127.0.0.1", 8080);
const char *env_ip = getenv("FASTSYNC_SERVER_IP");
const char *ip = env_ip ? env_ip : "127.0.0.1";
const char *env_port = getenv("FASTSYNC_SERVER_PORT");
int port = env_port ? atoi(env_port) : 8080;
client_connect(client, (char *)ip, port);
config_send(client->file_descriptor, config);
DirectoryScanner *scanner = directory_scanner_create(config->send_directory);
Chunk *current_chunk;
@@ -110,24 +126,15 @@ int send_files(Config *config) {
send_chunk(client, current_chunk, config);
chunk_destroy(current_chunk);
}
send_status(client->file_descriptor, FINISHED);
if (receive_status(client->file_descriptor) != OK)
send_status(client->file_descriptor, STATUS_FINISHED);
if (receive_status(client->file_descriptor) != STATUS_OK)
return -1;
printf("FINISHED");
directory_scanner_destroy(scanner);
client_disconnect(client);
client_delete(client);
return 0;
}
void handle_arg(char *argument_given, char *argument_to_set, bool *result,
char *message) {
if (strcmp(argument_given, argument_to_set) == 0) {
*result = true;
log_message(LOG_LEVEL_INFO, message);
}
}
int send_files_multithreaded(Config *config) {
PipelineContextSender *context =
pipeline_context_sender_create(config, queue_create(100, chunk_destroy),
@@ -151,40 +158,55 @@ int send_files_multithreaded(Config *config) {
return 0;
}
// int send_files_multiprocessed(Config *config) {
// int cores = sysconf(_SC_NPROCESSORS_ONLN);
// int pid = fork();
// if (pid == -1) {
// perror("Error Forking!");
// return 1;
// } else if (pid == 0) {
// }
// for (int i = 0; i < cores; i++) {
// int pid = fork();
// if (pid == -1) {
// perror("Error forking!");
// return 1;
// } else if (pid == 0) {
// }
// }
// return 0;
// }
void handle_arg(char *argument_given, char *argument_to_set, bool *result,
char *message) {
if (strcmp(argument_given, argument_to_set) == 0) {
*result = true;
log_message(LOG_LEVEL_INFO, message);
}
}
int main(int argc, char *argv[]) {
Config *config = config_create(
str_dup("1.0.0"), str_dup("/home/taptap/Nextcloud/Uni/moodle/MINT-Raum"),
str_dup("./data_copied"), false, false, false, false, 5, 1);
const char *env_source = getenv("FASTSYNC_SOURCE_DIR");
const char *env_dest = getenv("FASTSYNC_DEST_DIR");
const char *env_save = getenv("FASTSYNC_SAVE_TO_DISK");
char *source_dir =
env_source ? str_dup((char *)env_source)
: str_dup("/home/taptap/Nextcloud/Uni/moodle/B. Schnor "
"Konzepte Paralleler Programmierung, SoSe 2026");
char *dest_dir =
env_dest ? str_dup((char *)env_dest) : str_dup("./data_copied");
bool save_to_disk = false;
if (env_save &&
(strcmp(env_save, "true") == 0 || strcmp(env_save, "1") == 0)) {
save_to_disk = true;
}
Config *config = config_create(str_dup("1.0.0"), source_dir, dest_dir,
save_to_disk, false, false, false, 5, 20);
for (int i = 1; i < argc; i++) {
handle_arg(argv[i], "-m", &config->use_multithreading,
"Enabled Multithreading");
handle_arg(argv[i], "-s", &config->use_chunk_serialization,
"Enabled Chunk Serialization");
handle_arg(argv[i], "-c", &config->use_compression, "Enabled Compression");
if (strcmp(argv[i], "-c") == 0) {
config->use_compression = true;
log_message(LOG_LEVEL_INFO, "Enabled Compression");
if (i + 1 < argc) {
char *end_ptr;
int level = strtol(argv[i + 1], &end_ptr, 10);
if (*end_ptr == '\0') {
config->compression_level = level;
log_message(LOG_LEVEL_INFO, "Set Compression level to %d",
config->compression_level);
}
}
} else {
handle_arg(argv[i], "-m", &config->use_multithreading,
"Enabled Multithreading");
handle_arg(argv[i], "-s", &config->use_chunk_serialization,
"Enabled Chunk Serialization");
}
}
if (config->use_multithreading)
return send_files_multithreaded(config);
// else if (config->use_multiprocessing)
// return send_files_multiprocessed(config);
return send_files(config);
}