TapTap/FastSync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: split monolithic modules and extract shared components #1

Merged
TapTap merged 4 commits from cleanup into main 2026-07-16 11:58:38 +02:00
Conversation 0 Commits 4 Files Changed 34
+1119 -1142
34 changed files with 1119 additions and 1142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/client.c
-347
View File
@@ -1,347 +0,0 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <threads.h>
#include <unistd.h>
#include "chunk.h"
#include "config.h"
#include "data.h"
#include "file.h"
#include "log.h"
#include "multiprocessing.h"
#include "queue.h"
#include "scanner.h"
#include "socket.h"
#include "utils.h"
#include <dirent.h>
static char *server_host = "127.0.0.1";
static int server_port = 8080;
int send_chunk(Client *client, Chunk *chunk, Config *config) {
if (config->use_chunk_serialization) {
send_status(client->file_descriptor, STATUS_CHUNK);
Data *data;
if (config->use_compression) {
data = chunk_compress(chunk, config->compression_level, config->use_metadata);
} else {
for (int i = 0; i < chunk->element_count; i++)
file_load_data(chunk->items[i]);
data = chunk_serialize(chunk, config->use_metadata);
}
send_data(client->file_descriptor, data->data, data->size);
data_destroy(data);
} else if (config->use_sendfile && !config->use_compression) {
for (int i = 0; i < chunk->element_count; i++) {
send_status(client->file_descriptor, STATUS_NEXT);
file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata);
}
} else {
for (int i = 0; i < chunk->element_count; i++) {
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, config->use_metadata);
}
}
return 0;
}
int scan_directory_multithreaded(void *pipeline_context) {
PipelineContextSender *context = (PipelineContextSender *)pipeline_context;
mtx_lock(&context->mutex_scanner);
DirectoryScanner *scanner =
directory_scanner_create(context->config->send_directory, context->config->use_metadata);
mtx_unlock(&context->mutex_scanner);
Chunk *current_chunk;
while ((current_chunk = directory_scanner_next(scanner)) != NULL)
queue_enqueue_multithreaded(context->queue_scanner, current_chunk,
&context->mutex_scanner,
&context->condition_not_empty_scanner,
&context->condition_not_full_scanner);
mtx_lock(&context->mutex_scanner);
context->scanner_done = true;
cnd_signal(&context->condition_not_empty_scanner);
mtx_unlock(&context->mutex_scanner);
directory_scanner_destroy(scanner);
return thrd_success;
}
int load_files_multithreaded(void *pipeline_context) {
PipelineContextSender *context = (PipelineContextSender *)pipeline_context;
while (true) {
Chunk *chunk = queue_dequeue_multithreaded(
context->queue_scanner, &context->mutex_scanner,
&context->condition_not_empty_scanner,
&context->condition_not_full_scanner, &context->scanner_done);
if (chunk == NULL) {
mtx_lock(&context->mutex_loader);
context->loader_done = true;
cnd_signal(&context->condition_not_empty_loader);
mtx_unlock(&context->mutex_loader);
return thrd_success;
}
if (!context->config->use_sendfile) {
for (int i = 0; i < chunk->element_count; i++)
file_load_data(chunk->items[i]);
}
queue_enqueue_multithreaded(context->queue_loader, chunk,
&context->mutex_loader,
&context->condition_not_empty_loader,
&context->condition_not_full_loader);
}
}
int send_chunks_multithreaded(void *pipeline_context) {
PipelineContextSender *context = (PipelineContextSender *)pipeline_context;
Client *client;
if (context->config->transport == TRANSPORT_SSH) {
if (context->config->use_sendfile) {
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
return 1;
}
client = client_connect_ssh(context->config->ssh_destination);
} else {
client = client_create();
client_connect(client, server_host, server_port);
}
config_send(client->file_descriptor, context->config);
while (true) {
Chunk *current_chunk = queue_dequeue_multithreaded(
context->queue_loader, &context->mutex_loader,
&context->condition_not_empty_loader,
&context->condition_not_full_loader, &context->loader_done);
if (current_chunk == NULL) {
send_status(client->file_descriptor, STATUS_FINISHED);
int ok = receive_status(client->file_descriptor) == STATUS_OK;
client_disconnect(client);
client_delete(client);
return ok ? thrd_success : 1;
}
if (send_chunk(client, current_chunk, context->config) != 0) {
perror("Something unexpected happend while sending the chunk");
exit(EXIT_FAILURE);
}
chunk_destroy(current_chunk);
}
}
int send_files(Config *config) {
Client *client;
if (config->transport == TRANSPORT_SSH) {
if (config->use_sendfile) {
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
return 1;
}
client = client_connect_ssh(config->ssh_destination);
} else {
client = client_create();
client_connect(client, server_host, server_port);
}
config_send(client->file_descriptor, config);
DirectoryScanner *scanner = directory_scanner_create(config->send_directory, config->use_metadata);
Chunk *current_chunk;
while ((current_chunk = directory_scanner_next(scanner)) != NULL) {
if (!config->use_sendfile) {
for (int i = 0; i < current_chunk->element_count; i++)
file_load_data(current_chunk->items[i]);
}
send_chunk(client, current_chunk, config);
chunk_destroy(current_chunk);
}
send_status(client->file_descriptor, STATUS_FINISHED);
int ok = receive_status(client->file_descriptor) == STATUS_OK;
directory_scanner_destroy(scanner);
client_disconnect(client);
client_delete(client);
return ok ? 0 : -1;
}
int send_files_multithreaded(Config *config) {
PipelineContextSender *context =
pipeline_context_sender_create(config, queue_create(100, chunk_destroy),
queue_create(100, chunk_destroy));
thrd_t scanner, loader, sender;
if (thrd_create(&scanner, scan_directory_multithreaded, context) !=
thrd_success ||
thrd_create(&loader, load_files_multithreaded, context) != thrd_success ||
thrd_create(&sender, send_chunks_multithreaded, context) !=
thrd_success) {
perror("Error creating threads.\n");
return 1;
}
thrd_join(scanner, NULL);
thrd_join(loader, NULL);
thrd_join(sender, NULL);
pipeline_context_sender_destroy(context);
return 0;
}
static int is_remote_dest(const char *s) {
const char *colon = strchr(s, ':');
if (!colon) return 0;
if (colon == s) return 0;
for (const char *p = s; p < colon; p++) {
if (*p == '/') return 0;
}
return 1;
}
static void print_usage(void) {
printf("Usage:\n");
printf(" fastsync [options] <source> <destination>\n");
printf(" fastsync [options] --source-dir <src> --dest-dir <dst>\n");
printf("\n");
printf("Destination formats:\n");
printf(" user@host:/path SSH transport (rsync-style)\n");
printf(" host:/path SSH transport (current user)\n");
printf(" /local/path TCP transport (requires server on localhost:8080)\n");
printf("\n");
printf("Options:\n");
printf(" -c [level] Enable compression (level 1-22, default 5)\n");
printf(" -m Enable multithreading\n");
printf(" -s Enable chunk serialization\n");
printf(" -f Enable sendfile (TCP only, not with -c or -s)\n");
printf(" -v, --verbose Enable debug logging\n");
printf(" -M, --preserve Preserve file metadata\n");
printf(" --source-dir <path> Source directory\n");
printf(" --dest-dir <path> Destination directory\n");
printf(" --save-to-disk Write received files to disk\n");
printf(" --server-host <ip> Server IP address (default: 127.0.0.1)\n");
printf(" --server-port <n> Server port (default: 8080)\n");
printf(" --help Show this help\n");
}
int main(int argc, char *argv[]) {
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");
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"), NULL, NULL,
save_to_disk, false, false, false, false, 5, 20, false);
int positional_args[2];
int positional_count = 0;
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "--help") == 0) {
print_usage();
return 0;
} else 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);
i++;
}
}
} else if (strcmp(argv[i], "--source-dir") == 0 && i + 1 < argc) {
free(config->send_directory);
config->send_directory = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--dest-dir") == 0 && i + 1 < argc) {
free(config->receive_root_directory);
config->receive_root_directory = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--save-to-disk") == 0) {
config->save_to_disk = true;
} else if (strcmp(argv[i], "-M") == 0 || strcmp(argv[i], "--preserve") == 0) {
config->use_metadata = true;
log_message(LOG_LEVEL_INFO, "Enabled metadata preservation");
} else if (strcmp(argv[i], "-f") == 0 || strcmp(argv[i], "--sendfile") == 0) {
config->use_sendfile = true;
log_message(LOG_LEVEL_INFO, "Enabled sendfile");
} else if (strcmp(argv[i], "-m") == 0) {
config->use_multithreading = true;
log_message(LOG_LEVEL_INFO, "Enabled Multithreading");
} else if (strcmp(argv[i], "-s") == 0) {
config->use_chunk_serialization = true;
log_message(LOG_LEVEL_INFO, "Enabled Chunk Serialization");
} else if (strcmp(argv[i], "--server-host") == 0 && i + 1 < argc) {
free(server_host);
server_host = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) {
server_port = atoi(argv[++i]);
} else if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--verbose") == 0) {
set_log_level(LOG_LEVEL_DEBUG);
} else if (argv[i][0] == '-') {
fprintf(stderr, "Unknown option: %s\n", argv[i]);
print_usage();
return 1;
} else {
if (positional_count < 2)
positional_args[positional_count++] = i;
else {
fprintf(stderr, "Unexpected argument: %s\n", argv[i]);
print_usage();
return 1;
}
}
}
if (positional_count == 2) {
free(config->send_directory);
free(config->receive_root_directory);
config->send_directory = str_dup(argv[positional_args[0]]);
config->receive_root_directory = str_dup(argv[positional_args[1]]);
config->save_to_disk = true;
if (is_remote_dest(config->receive_root_directory)) {
config->transport = TRANSPORT_SSH;
config->ssh_destination = str_dup(config->receive_root_directory);
char *colon = strchr(config->receive_root_directory, ':');
char *path = str_dup(colon + 1);
free(config->receive_root_directory);
config->receive_root_directory = path;
}
} else if (positional_count == 1) {
fprintf(stderr, "Error: missing destination argument\n");
print_usage();
return 1;
} else {
if (!config->send_directory && env_source)
config->send_directory = str_dup((char *)env_source);
if (!config->receive_root_directory && env_dest)
config->receive_root_directory = str_dup((char *)env_dest);
}
if (!config->send_directory || !config->receive_root_directory) {
fprintf(stderr, "Error: source and destination directories are required\n");
print_usage();
return 1;
}
if (config->use_sendfile && (config->use_chunk_serialization || config->use_compression)) {
fprintf(stderr, "Error: -f/--sendfile cannot be combined with -c (compression) or -s (chunk serialization)\n");
return 1;
}
if (config->transport == TRANSPORT_SSH && config->use_sendfile) {
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
return 1;
}
if (config->use_multithreading)
return send_files_multithreaded(config);
return send_files(config);
}
src/client/client_cli.c
+150
View File
@@ -0,0 +1,150 @@
#include "client_send.h"
#include "config.h"
#include "log.h"
#include "utils.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
char *server_host = "127.0.0.1";
int server_port = 8080;
static void print_usage(void) {
printf("Usage:\n");
printf(" fastsync [options] <source> <destination>\n");
printf(" fastsync [options] --source-dir <src> --dest-dir <dst>\n");
printf("\n");
printf("Destination formats:\n");
printf(" user@host:/path SSH transport (rsync-style)\n");
printf(" host:/path SSH transport (current user)\n");
printf(" /local/path TCP transport (requires server on localhost:8080)\n");
printf("\n");
printf("Options:\n");
printf(" -c [level] Enable compression (level 1-22, default 5)\n");
printf(" -m Enable multithreading\n");
printf(" -s Enable chunk serialization\n");
printf(" -f Enable sendfile (TCP only, not with -c or -s)\n");
printf(" -v, --verbose Enable debug logging\n");
printf(" -M, --preserve Preserve file metadata\n");
printf(" --source-dir <path> Source directory\n");
printf(" --dest-dir <path> Destination directory\n");
printf(" --save-to-disk Write received files to disk\n");
printf(" --server-host <ip> Server IP address (default: 127.0.0.1)\n");
printf(" --server-port <n> Server port (default: 8080)\n");
printf(" --help Show this help\n");
}
int main(int argc, char *argv[]) {
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");
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"), NULL, NULL,
save_to_disk, false, false, false, false, 5, 20, false);
int positional_args[2];
int positional_count = 0;
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "--help") == 0) {
print_usage();
return 0;
} else 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);
i++;
}
}
} else if (strcmp(argv[i], "--source-dir") == 0 && i + 1 < argc) {
free(config->send_directory);
config->send_directory = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--dest-dir") == 0 && i + 1 < argc) {
free(config->receive_root_directory);
config->receive_root_directory = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--save-to-disk") == 0) {
config->save_to_disk = true;
} else if (strcmp(argv[i], "-M") == 0 || strcmp(argv[i], "--preserve") == 0) {
config->use_metadata = true;
log_message(LOG_LEVEL_INFO, "Enabled metadata preservation");
} else if (strcmp(argv[i], "-f") == 0 || strcmp(argv[i], "--sendfile") == 0) {
config->use_sendfile = true;
log_message(LOG_LEVEL_INFO, "Enabled sendfile");
} else if (strcmp(argv[i], "-m") == 0) {
config->use_multithreading = true;
log_message(LOG_LEVEL_INFO, "Enabled Multithreading");
} else if (strcmp(argv[i], "-s") == 0) {
config->use_chunk_serialization = true;
log_message(LOG_LEVEL_INFO, "Enabled Chunk Serialization");
} else if (strcmp(argv[i], "--server-host") == 0 && i + 1 < argc) {
free(server_host);
server_host = str_dup(argv[++i]);
} else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) {
server_port = atoi(argv[++i]);
} else if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--verbose") == 0) {
set_log_level(LOG_LEVEL_DEBUG);
} else if (argv[i][0] == '-') {
fprintf(stderr, "Unknown option: %s\n", argv[i]);
print_usage();
return 1;
} else {
if (positional_count < 2)
positional_args[positional_count++] = i;
else {
fprintf(stderr, "Unexpected argument: %s\n", argv[i]);
print_usage();
return 1;
}
}
}
if (positional_count == 2) {
free(config->send_directory);
free(config->receive_root_directory);
config->send_directory = str_dup(argv[positional_args[0]]);
config->receive_root_directory = str_dup(argv[positional_args[1]]);
config->save_to_disk = true;
config_parse_ssh_dest(config);
} else if (positional_count == 1) {
fprintf(stderr, "Error: missing destination argument\n");
print_usage();
return 1;
} else {
if (!config->send_directory && env_source)
config->send_directory = str_dup((char *)env_source);
if (!config->receive_root_directory && env_dest)
config->receive_root_directory = str_dup((char *)env_dest);
}
if (!config->send_directory || !config->receive_root_directory) {
fprintf(stderr, "Error: source and destination directories are required\n");
print_usage();
return 1;
}
if (config->use_sendfile && (config->use_chunk_serialization || config->use_compression)) {
fprintf(stderr, "Error: -f/--sendfile cannot be combined with -c (compression) or -s (chunk serialization)\n");
return 1;
}
if (config->transport == TRANSPORT_SSH && config->use_sendfile) {
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
return 1;
}
if (config->use_multithreading)
return send_files_multithreaded(config);
return send_files(config);
}
src/client/client_send.c
+180
View File
@@ -0,0 +1,180 @@
#include "client_send.h"
#include "chunk.h"
#include "config.h"
#include "data.h"
#include "file.h"
#include "log.h"
#include "multiprocessing.h"
#include "protocol.h"
#include "queue.h"
#include "scanner.h"
#include "transport_tcp.h"
#include "transport_ssh.h"
#include "utils.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <threads.h>
int send_chunk(Client *client, Chunk *chunk, Config *config) {
if (config->use_chunk_serialization) {
send_status(client->file_descriptor, STATUS_CHUNK);
Data *data;
if (config->use_compression) {
data = chunk_compress(chunk, config->compression_level, config->use_metadata);
} else {
data = chunk_serialize(chunk, config->use_metadata);
}
send_data(client->file_descriptor, data);
data_destroy(data);
} else if (config->use_sendfile && !config->use_compression) {
for (int i = 0; i < chunk->element_count; i++) {
send_status(client->file_descriptor, STATUS_NEXT);
file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata);
}
} else {
for (int i = 0; i < chunk->element_count; i++) {
send_status(client->file_descriptor, STATUS_NEXT);
file_send_single_calls(chunk->items[i], client->file_descriptor,
config->use_metadata,
config->use_compression ? config->compression_level : 0);
}
}
return 0;
}
static int send_chunks_multithreaded(void *pipeline_context) {
PipelineContextSender *context = (PipelineContextSender *)pipeline_context;
Client *client;
if (context->config->transport == TRANSPORT_SSH) {
if (context->config->use_sendfile) {
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
return 1;
}
client = client_connect_ssh(context->config->ssh_destination);
} else {
client = client_create();
client_connect(client, server_host, server_port);
}
config_send(client->file_descriptor, context->config);
while (true) {
Chunk *current_chunk = queue_dequeue_multithreaded(
context->queue_loader, &context->mutex_loader,
&context->condition_not_empty_loader,
&context->condition_not_full_loader, &context->loader_done);
if (current_chunk == NULL) {
send_status(client->file_descriptor, STATUS_FINISHED);
int ok = receive_status(client->file_descriptor) == STATUS_OK;
client_disconnect(client);
client_delete(client);
return ok ? thrd_success : thrd_error;
}
if (send_chunk(client, current_chunk, context->config) != 0) {
perror("Something unexpected happend while sending the chunk");
exit(EXIT_FAILURE);
}
chunk_destroy(current_chunk);
}
}
static int scan_directory_multithreaded(void *pipeline_context) {
PipelineContextSender *context = (PipelineContextSender *)pipeline_context;
mtx_lock(&context->mutex_scanner);
DirectoryScanner *scanner =
directory_scanner_create(context->config->send_directory, context->config->use_metadata);
mtx_unlock(&context->mutex_scanner);
Chunk *current_chunk;
while ((current_chunk = directory_scanner_next(scanner)) != NULL)
queue_enqueue_multithreaded(context->queue_scanner, current_chunk,
&context->mutex_scanner,
&context->condition_not_empty_scanner,
&context->condition_not_full_scanner);
mtx_lock(&context->mutex_scanner);
context->scanner_done = true;
cnd_signal(&context->condition_not_empty_scanner);
mtx_unlock(&context->mutex_scanner);
directory_scanner_destroy(scanner);
return thrd_success;
}
static int load_files_multithreaded(void *pipeline_context) {
PipelineContextSender *context = (PipelineContextSender *)pipeline_context;
while (true) {
Chunk *chunk = queue_dequeue_multithreaded(
context->queue_scanner, &context->mutex_scanner,
&context->condition_not_empty_scanner,
&context->condition_not_full_scanner, &context->scanner_done);
if (chunk == NULL) {
mtx_lock(&context->mutex_loader);
context->loader_done = true;
cnd_signal(&context->condition_not_empty_loader);
mtx_unlock(&context->mutex_loader);
return thrd_success;
}
if (!context->config->use_sendfile) {
for (int i = 0; i < chunk->element_count; i++)
file_load_data(chunk->items[i]);
}
queue_enqueue_multithreaded(context->queue_loader, chunk,
&context->mutex_loader,
&context->condition_not_empty_loader,
&context->condition_not_full_loader);
}
}
int send_files(Config *config) {
Client *client;
if (config->transport == TRANSPORT_SSH) {
if (config->use_sendfile) {
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
return 1;
}
client = client_connect_ssh(config->ssh_destination);
} else {
client = client_create();
client_connect(client, server_host, server_port);
}
config_send(client->file_descriptor, config);
DirectoryScanner *scanner = directory_scanner_create(config->send_directory, config->use_metadata);
Chunk *current_chunk;
while ((current_chunk = directory_scanner_next(scanner)) != NULL) {
if (!config->use_sendfile) {
for (int i = 0; i < current_chunk->element_count; i++)
file_load_data(current_chunk->items[i]);
}
send_chunk(client, current_chunk, config);
chunk_destroy(current_chunk);
}
send_status(client->file_descriptor, STATUS_FINISHED);
int ok = receive_status(client->file_descriptor) == STATUS_OK;
directory_scanner_destroy(scanner);
client_disconnect(client);
client_delete(client);
return ok ? 0 : -1;
}
int send_files_multithreaded(Config *config) {
PipelineContextSender *context =
pipeline_context_sender_create(config, queue_create(100, chunk_destroy),
queue_create(100, chunk_destroy));
thrd_t scanner, loader, sender;
if (thrd_create(&scanner, scan_directory_multithreaded, context) !=
thrd_success ||
thrd_create(&loader, load_files_multithreaded, context) != thrd_success ||
thrd_create(&sender, send_chunks_multithreaded, context) !=
thrd_success) {
perror("Error creating threads.\n");
return 1;
}
thrd_join(scanner, NULL);
thrd_join(loader, NULL);
thrd_join(sender, NULL);
pipeline_context_sender_destroy(context);
return 0;
}
src/client/client_send.h
+15
View File
@@ -0,0 +1,15 @@
#ifndef CLIENT_SEND_H
#define CLIENT_SEND_H
#include "chunk.h"
#include "config.h"
#include "transport_tcp.h"
extern char *server_host;
extern int server_port;
int send_chunk(Client *client, Chunk *chunk, Config *config);
int send_files(Config *config);
int send_files_multithreaded(Config *config);
#endif
src/server/server.c
+5 -95
View File
@@ -1,109 +1,19 @@
#include "chunk.h"
#include "compression.h"
#include "config.h"
#include "data.h"
#include "file.h"
#include "log.h"
#include "metadata.h"
#include "multiprocessing.h"
#include "protocol.h"
#include "queue.h"
#include "socket.h"
#include "transport_tcp.h"
#include "unistd.h"
#include "utils.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <threads.h>
File *file_receive(Config *config, int file_descriptor) {
char *path = (char *)receive_str(file_descriptor);
File *file = file_create(path);
free(path);
if (config->use_metadata)
file->metadata = file_receive_metadata(file_descriptor);
Data *file_data = receive_data(file_descriptor);
if (config->use_compression) {
Data *file_data_uncompressed = data_decompress(file_data);
data_destroy(file_data);
file_data = file_data_uncompressed;
}
data_destroy(file->data);
file->data = file_data;
return file;
}
static void receive_chunk_enqueue(int file_descriptor, Config *config,
PipelineContextReceiver *context) {
(void)config;
Data *chunk_data = receive_data(file_descriptor);
Data *data_to_process = chunk_data;
if (config->use_compression) {
data_to_process = data_decompress(chunk_data);
data_destroy(chunk_data);
}
Chunk *chunk = chunk_deserialize(data_to_process, config->use_metadata);
data_destroy(data_to_process);
for (int i = 0; i < chunk->element_count; i++) {
File *file = chunk->items[i];
chunk->items[i] = NULL;
queue_enqueue_multithreaded(context->queue, file, &context->mutex,
&context->condition_not_empty,
&context->condition_not_full);
}
chunk_destroy(chunk);
}
int receive_thread(void *pipeline_context) {
PipelineContextReceiver *context =
(PipelineContextReceiver *)pipeline_context;
mtx_lock(&context->mutex);
int file_descriptor = context->file_descriptor;
Config *config = context->config;
mtx_unlock(&context->mutex);
Status status = receive_status(file_descriptor);
while (status == STATUS_NEXT || status == STATUS_CHUNK) {
if (status == STATUS_CHUNK) {
receive_chunk_enqueue(file_descriptor, config, context);
} else {
File *file = file_receive(config, file_descriptor);
queue_enqueue_multithreaded(context->queue, file, &context->mutex,
&context->condition_not_empty,
&context->condition_not_full);
}
status = receive_status(file_descriptor);
}
mtx_lock(&context->mutex);
context->receiver_done = true;
cnd_signal(&context->condition_not_empty);
mtx_unlock(&context->mutex);
return thrd_success;
}
int write_thread(void *pipeline_context) {
PipelineContextReceiver *context =
(PipelineContextReceiver *)pipeline_context;
mtx_lock(&context->mutex);
bool save_to_disk = context->config->save_to_disk;
char *root_directory = str_dup(context->config->receive_root_directory);
mtx_unlock(&context->mutex);
while (true) {
File *file = queue_dequeue_multithreaded(
context->queue, &context->mutex, &context->condition_not_empty,
&context->condition_not_full, &context->receiver_done);
if (file == NULL) {
free(root_directory);
return thrd_success;
}
if (save_to_disk) {
char *disk_path = path_cat(root_directory, file->path);
to_disk(disk_path, file->data->data, file->data->size);
file_restore_metadata(disk_path, file->metadata);
free(disk_path);
}
file_destroy(file);
}
}
int receive_files(Config *config, int file_descriptor) {
Status status = receive_status(file_descriptor);
@@ -180,6 +90,6 @@ int main(int argc, char *argv[]) {
}
Server *server = server_create(8080);
server_listen(server, handler);
server_delete(server);
server_delete(&server);
return 0;
}
src/shared/chunk.c
+3 -104
View File
@@ -1,17 +1,15 @@
#include <dirent.h>
#include <libgen.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "chunk.h"
#include "array_list.h"
#include "chunk.h"
#include "compression.h"
#include "data.h"
#include "file.h"
#include "log.h"
#define FILE_METADATA_WIRE_SIZE (sizeof(mode_t) + sizeof(uid_t) + sizeof(gid_t) + sizeof(time_t) + sizeof(long))
#include "metadata.h"
Chunk *chunk_create(File **items, int element_count) {
Chunk *chunk = (Chunk *)malloc(sizeof(Chunk));
@@ -49,90 +47,6 @@ void chunk_destroy(void *item) {
free(chunk);
}
void chunk_print(void *item) {
if (item == NULL)
return;
Chunk *chunk = (Chunk *)item;
for (int i = 0; i < chunk->element_count; ++i)
if (chunk->items[i] != NULL)
file_print(chunk->items[i]);
}
static void metadata_to_buf(char **buf, FileMetadata *m) {
int present = (m != NULL) ? 1 : 0;
memcpy(*buf, &present, sizeof(int));
*buf += sizeof(int);
if (m == NULL)
return;
memcpy(*buf, &m->mode, sizeof(mode_t)); *buf += sizeof(mode_t);
memcpy(*buf, &m->uid, sizeof(uid_t)); *buf += sizeof(uid_t);
memcpy(*buf, &m->gid, sizeof(gid_t)); *buf += sizeof(gid_t);
memcpy(*buf, &m->mtime_sec, sizeof(time_t)); *buf += sizeof(time_t);
memcpy(*buf, &m->mtime_nsec, sizeof(long)); *buf += sizeof(long);
}
static FileMetadata *metadata_from_buf(char **buf) {
int present;
memcpy(&present, *buf, sizeof(int));
*buf += sizeof(int);
if (!present)
return NULL;
FileMetadata *m = malloc(sizeof(FileMetadata));
memcpy(&m->mode, *buf, sizeof(mode_t)); *buf += sizeof(mode_t);
memcpy(&m->uid, *buf, sizeof(uid_t)); *buf += sizeof(uid_t);
memcpy(&m->gid, *buf, sizeof(gid_t)); *buf += sizeof(gid_t);
memcpy(&m->mtime_sec, *buf, sizeof(time_t)); *buf += sizeof(time_t);
memcpy(&m->mtime_nsec, *buf, sizeof(long)); *buf += sizeof(long);
return m;
}
static unsigned long long per_file_chunk_format_size(File *file) {
return sizeof(int) + strlen(file->path) + sizeof(int) +
(file->metadata ? FILE_METADATA_WIRE_SIZE : 0) +
sizeof(unsigned long long) + file->data->size;
}
Data *chunk_format(Chunk *chunk) {
unsigned long long buffer_size = 0;
for (int i = 0; i < chunk->element_count; ++i) {
buffer_size += per_file_chunk_format_size(chunk->items[i]);
}
char *data = malloc(buffer_size);
if (data == NULL) {
perror("Could not allocate data for ChunkFormated!");
exit(EXIT_FAILURE);
}
char *current_data_pointer = data;
for (int i = 0; i < chunk->element_count; ++i) {
File *file = chunk->items[i];
// add path len
int path_length = (int)strlen(file->path);
memcpy(current_data_pointer, &path_length, sizeof(int));
current_data_pointer += sizeof(int);
// add path
memcpy(current_data_pointer, file->path, path_length);
current_data_pointer += path_length;
// add metadata
metadata_to_buf(&current_data_pointer, file->metadata);
// add file data len
unsigned long long file_length = file->data->size;
memcpy(current_data_pointer, &file_length, sizeof(unsigned long long));
current_data_pointer += sizeof(unsigned long long);
// add file data
if (file->data->data == NULL) {
file_load_data(file);
}
memcpy(current_data_pointer, file->data->data, file_length);
current_data_pointer += file_length;
}
if (current_data_pointer - data != (long)(long)buffer_size) {
perror("Buffer of Chunk wasn't filled enough!");
exit(EXIT_FAILURE);
}
return chunk_data_create(data, buffer_size);
}
static unsigned long long per_file_serialize_size(File *file, bool use_metadata) {
return sizeof(size_t) + strlen(file->path) +
(use_metadata ? sizeof(int) + (file->metadata ? FILE_METADATA_WIRE_SIZE : 0) : 0) +
@@ -284,19 +198,4 @@ Chunk *chunk_decompress(Data *compressed_data, bool use_metadata) {
return chunk;
}
Data *chunk_data_create(void *data, unsigned long long data_size) {
Data *chunk_formated = malloc(sizeof(Data));
if (chunk_formated == NULL) {
perror("Could not allocate memory for ChunkFormated");
exit(EXIT_FAILURE);
}
chunk_formated->data = data;
chunk_formated->size = data_size;
return chunk_formated;
}
void chunk_data_delete(void *chunk) {
Data *chunk_data = (Data *)chunk;
free(chunk_data->data);
free(chunk_data);
}
src/shared/chunk.h
+1 -4
View File
@@ -15,13 +15,10 @@ typedef struct {
Chunk *chunk_create(File **items, int element_count);
void chunk_destroy(void *chunk);
void chunk_print(void *chunk);
Data *chunk_format(Chunk *chunk);
Data *chunk_serialize(Chunk *chunk, bool use_metadata);
Chunk *chunk_deserialize(Data *data, bool use_metadata);
Data *chunk_compress(Chunk *chunk, int compression_level, bool use_metadata);
Chunk *chunk_decompress(Data *compressed_data, bool use_metadata);
Data *chunk_data_create(void *data, unsigned long long data_size);
void chunk_data_delete(void *chunk);
#endif
src/shared/compression.c
+46
View File
@@ -0,0 +1,46 @@
#include "compression.h"
#include "data.h"
#include "log.h"
#include "stdlib.h"
#include "zstd.h"
Data *data_compress(Data *data_to_compress, int compression_level) {
log_message(LOG_LEVEL_DEBUG, "Starting to compress data");
Data *compressed_data =
data_create_empty(ZSTD_compressBound(data_to_compress->size));
compressed_data->size = ZSTD_compress(
compressed_data->data, compressed_data->size, data_to_compress->data,
data_to_compress->size, compression_level);
if (ZSTD_isError(compressed_data->size)) {
log_message(LOG_LEVEL_ERROR, "Compression failed: %s",
ZSTD_getErrorName(compressed_data->size));
exit(EXIT_FAILURE);
}
log_message(LOG_LEVEL_DEBUG, "Data succesfully compressed from %zu to %zu",
data_to_compress->size, compressed_data->size);
return compressed_data;
}
Data *data_decompress(Data *compressed_data) {
log_message(LOG_LEVEL_DEBUG, "Start to decompress data");
Data *uncompressed_data = data_create_empty(
ZSTD_getFrameContentSize(compressed_data->data, compressed_data->size));
if (ZSTD_isError(uncompressed_data->size)) {
log_message(LOG_LEVEL_ERROR, "Decompression failed: %s",
ZSTD_getErrorName(uncompressed_data->size));
exit(EXIT_FAILURE);
}
uncompressed_data->size =
ZSTD_decompress(uncompressed_data->data, uncompressed_data->size,
compressed_data->data, compressed_data->size);
if (ZSTD_isError(uncompressed_data->size)) {
log_message(LOG_LEVEL_ERROR, "Decompression failed: %s",
ZSTD_getErrorName(uncompressed_data->size));
exit(EXIT_FAILURE);
}
log_message(LOG_LEVEL_DEBUG, "Decompressed data successfully");
return uncompressed_data;
}
src/shared/compression.h
+9
View File
@@ -0,0 +1,9 @@
#ifndef COMPRESSION_H
#define COMPRESSION_H
#include "data.h"
Data *data_compress(Data *data_to_compress, int compression_level);
Data *data_decompress(Data *compressed_data);
#endif
src/shared/config.c
+24 -1
View File
@@ -1,8 +1,10 @@
#include "config.h"
#include "socket.h"
#include "protocol.h"
#include "utils.h"
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
Config *config_create(char *version, char *send_directory,
char *receive_directory, bool save_to_disk,
@@ -27,6 +29,27 @@ Config *config_create(char *version, char *send_directory,
return config;
}
bool is_remote_dest(const char *s) {
if (s == NULL) return false;
const char *colon = strchr(s, ':');
if (colon == NULL) return false;
if (colon == s) return false;
for (const char *p = s; p < colon; p++) {
if (*p == '/') return false;
}
return true;
}
void config_parse_ssh_dest(Config *config) {
if (!is_remote_dest(config->receive_root_directory)) return;
config->transport = TRANSPORT_SSH;
config->ssh_destination = str_dup(config->receive_root_directory);
char *colon = strchr(config->receive_root_directory, ':');
char *path = str_dup(colon + 1);
free(config->receive_root_directory);
config->receive_root_directory = path;
}
void config_delete(Config *config) {
free(config->version);
free(config->send_directory);
src/shared/config.h
+2 -1
View File
@@ -17,7 +17,6 @@ typedef struct Config {
bool use_chunk_serialization;
bool use_compression;
bool use_sendfile;
bool use_single_send_per_file;
bool use_metadata;
int compression_level;
int num_connections;
@@ -33,5 +32,7 @@ Config *config_create(char *version, char *send_directory,
void config_delete(Config *config);
void config_send(int file_descriptor, Config *config);
Config *config_receive(int file_descriptor);
bool is_remote_dest(const char *s);
void config_parse_ssh_dest(Config *config);
#endif
src/shared/data.c
-43
View File
@@ -1,8 +1,6 @@
#include "data.h"
#include "log.h"
#include "stdlib.h"
#include "zstd.h"
Data *data_create_empty(size_t data_size) {
void *data = malloc(data_size);
@@ -40,44 +38,3 @@ void data_destroy(Data *data) {
free(data->data);
free(data);
}
Data *data_compress(Data *data_to_compress, int compression_level) {
log_message(LOG_LEVEL_DEBUG, "Starting to compress data");
Data *compressed_data =
data_create_empty(ZSTD_compressBound(data_to_compress->size));
compressed_data->size = ZSTD_compress(
compressed_data->data, compressed_data->size, data_to_compress->data,
data_to_compress->size, compression_level);
if (ZSTD_isError(compressed_data->size)) {
log_message(LOG_LEVEL_ERROR, "Compression failed: %s",
ZSTD_getErrorName(compressed_data->size));
exit(EXIT_FAILURE);
}
log_message(LOG_LEVEL_DEBUG, "Data succesfully compressed from %zu to %zu",
data_to_compress->size, compressed_data->size);
return compressed_data;
}
Data *data_decompress(Data *compressed_data) {
log_message(LOG_LEVEL_DEBUG, "Start to decompress data");
Data *uncompressed_data = data_create_empty(
ZSTD_getFrameContentSize(compressed_data->data, compressed_data->size));
if (ZSTD_isError(uncompressed_data->size)) {
log_message(LOG_LEVEL_ERROR, "Decompression failed: %s",
ZSTD_getErrorName(uncompressed_data->size));
exit(EXIT_FAILURE);
}
uncompressed_data->size =
ZSTD_decompress(uncompressed_data->data, uncompressed_data->size,
compressed_data->data, compressed_data->size);
if (ZSTD_isError(uncompressed_data->size)) {
log_message(LOG_LEVEL_ERROR, "Decompression failed: %s",
ZSTD_getErrorName(uncompressed_data->size));
exit(EXIT_FAILURE);
}
log_message(LOG_LEVEL_DEBUG, "Decompressed data successfully");
return uncompressed_data;
}
src/shared/data.h
-2
View File
@@ -12,7 +12,5 @@ Data *data_create_empty(size_t data_size);
Data *data_create_reserve(size_t size);
Data *data_create(void *data, size_t data_size);
void data_destroy(Data *data);
Data *data_compress(Data *data_to_compress, int compression_level);
Data *data_decompress(Data *compressed_data);
#endif
src/shared/file.c
+46 -40
View File
@@ -6,13 +6,17 @@
#include <stdlib.h>
#include <string.h>
#include <sys/sendfile.h>
#include <sys/stat.h>
#include <unistd.h>
#include <zstd.h>
#include "compression.h"
#include "config.h"
#include "data.h"
#include "file.h"
#include "log.h"
#include "socket.h"
#include "metadata.h"
#include "protocol.h"
#include "utils.h"
File *file_create(const char *path) {
File *file = (File *)malloc(sizeof(File));
@@ -80,7 +84,6 @@ void file_load_data(File *file) {
exit(EXIT_FAILURE);
}
}
printf("%ld is file big", file->data->size);
size_t bytes_read = file_content_to_buffer(file);
if (bytes_read != file->data->size) {
log_message(STATUS_ERROR, "Didnt read expected amount of bytes from file");
@@ -88,47 +91,31 @@ void file_load_data(File *file) {
}
}
void file_print(void *item) {
if (item == NULL)
return;
printf("%s\n", ((File *)item)->path);
}
static void metadata_send(int file_descriptor, FileMetadata *m) {
if (m == NULL) {
int zero = 0;
send_n_data(file_descriptor, &zero, sizeof(int));
return;
void file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level) {
if (compression_level > 0) {
Data *compressed_data = data_compress(file->data, compression_level);
data_destroy(file->data);
file->data = compressed_data;
}
int present = 1;
send_n_data(file_descriptor, &present, sizeof(int));
send_n_data(file_descriptor, &m->mode, sizeof(mode_t));
send_n_data(file_descriptor, &m->uid, sizeof(uid_t));
send_n_data(file_descriptor, &m->gid, sizeof(gid_t));
send_n_data(file_descriptor, &m->mtime_sec, sizeof(time_t));
send_n_data(file_descriptor, &m->mtime_nsec, sizeof(long));
}
FileMetadata *file_receive_metadata(int file_descriptor) {
int present;
receive_n_data(file_descriptor, &present, sizeof(int));
if (!present)
return NULL;
FileMetadata *m = malloc(sizeof(FileMetadata));
receive_n_data(file_descriptor, &m->mode, sizeof(mode_t));
receive_n_data(file_descriptor, &m->uid, sizeof(uid_t));
receive_n_data(file_descriptor, &m->gid, sizeof(gid_t));
receive_n_data(file_descriptor, &m->mtime_sec, sizeof(time_t));
receive_n_data(file_descriptor, &m->mtime_nsec, sizeof(long));
return m;
}
void file_send_single_calls(File *file, int file_descriptor, bool use_metadata) {
send_str(file_descriptor, file->path);
if (use_metadata)
metadata_send(file_descriptor, file->metadata);
printf("Sending File: %ld", file->data->size);
send_data(file_descriptor, file->data->data, file->data->size);
send_data(file_descriptor, file->data);
}
void to_disk(const char *path, const void *data, unsigned long long data_size) {
char *directory = str_dup(path);
char *dir_to_free = directory;
directory = dirname(directory);
mkdir_r(directory);
FILE *file_pointer = fopen(path, "wb");
if (file_pointer == NULL) {
perror("Could not open File");
exit(EXIT_FAILURE);
}
fwrite(data, 1, data_size, file_pointer);
fclose(file_pointer);
free(dir_to_free);
}
void file_send_sendfile(File *file, int file_descriptor, bool use_metadata) {
@@ -158,6 +145,23 @@ void file_send_sendfile(File *file, int file_descriptor, bool use_metadata) {
close(fd);
}
File *file_receive(Config *config, int file_descriptor) {
char *path = (char *)receive_str(file_descriptor);
File *file = file_create(path);
free(path);
if (config->use_metadata)
file->metadata = metadata_receive(file_descriptor);
Data *file_data = receive_data(file_descriptor);
if (config->use_compression) {
Data *file_data_uncompressed = data_decompress(file_data);
data_destroy(file_data);
file_data = file_data_uncompressed;
}
data_destroy(file->data);
file->data = file_data;
return file;
}
size_t file_content_to_buffer(File *file) {
FILE *file_pointer = fopen(file->path, "rb");
if (file_pointer == NULL) {
@@ -173,3 +177,5 @@ size_t file_content_to_buffer(File *file) {
fclose(file_pointer);
return bytes_read;
}
src/shared/file.h
+4 -3
View File
@@ -1,6 +1,7 @@
#ifndef FILE_H
#define FILE_H
#include "config.h"
#include "data.h"
#include <stdbool.h>
#include <sys/stat.h>
@@ -22,12 +23,12 @@ typedef struct {
File *file_create(const char *path);
void file_destroy(void *item);
void file_load_data(File *file);
void file_print(void *item);
void file_send_single_calls(File *file, int file_descriptor, bool use_metadata);
File *file_receive(Config *config, int file_descriptor);
void file_send_single_calls(File *file, int file_descriptor, bool use_metadata, int compression_level);
void file_send_sendfile(File *file, int file_descriptor, bool use_metadata);
size_t file_content_to_buffer(File *file);
FileMetadata *file_metadata_create(struct stat *stats);
void file_metadata_destroy(void *metadata);
FileMetadata *file_receive_metadata(int file_descriptor);
void to_disk(const char *path, const void *data, unsigned long long data_size);
#endif
src/shared/metadata.c
+82
View File
@@ -0,0 +1,82 @@
#include "metadata.h"
#include "file.h"
#include "protocol.h"
#include <fcntl.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>
#define FILE_METADATA_WIRE_SIZE (sizeof(mode_t) + sizeof(uid_t) + sizeof(gid_t) + sizeof(time_t) + sizeof(long))
void metadata_to_buf(char **buf, FileMetadata *m) {
int present = (m != NULL) ? 1 : 0;
memcpy(*buf, &present, sizeof(int));
*buf += sizeof(int);
if (m == NULL)
return;
memcpy(*buf, &m->mode, sizeof(mode_t)); *buf += sizeof(mode_t);
memcpy(*buf, &m->uid, sizeof(uid_t)); *buf += sizeof(uid_t);
memcpy(*buf, &m->gid, sizeof(gid_t)); *buf += sizeof(gid_t);
memcpy(*buf, &m->mtime_sec, sizeof(time_t)); *buf += sizeof(time_t);
memcpy(*buf, &m->mtime_nsec, sizeof(long)); *buf += sizeof(long);
}
FileMetadata *metadata_from_buf(char **buf) {
int present;
memcpy(&present, *buf, sizeof(int));
*buf += sizeof(int);
if (!present)
return NULL;
FileMetadata *m = malloc(sizeof(FileMetadata));
memcpy(&m->mode, *buf, sizeof(mode_t)); *buf += sizeof(mode_t);
memcpy(&m->uid, *buf, sizeof(uid_t)); *buf += sizeof(uid_t);
memcpy(&m->gid, *buf, sizeof(gid_t)); *buf += sizeof(gid_t);
memcpy(&m->mtime_sec, *buf, sizeof(time_t)); *buf += sizeof(time_t);
memcpy(&m->mtime_nsec, *buf, sizeof(long)); *buf += sizeof(long);
return m;
}
void metadata_send(int file_descriptor, FileMetadata *m) {
if (m == NULL) {
int zero = 0;
send_n_data(file_descriptor, &zero, sizeof(int));
return;
}
int present = 1;
send_n_data(file_descriptor, &present, sizeof(int));
send_n_data(file_descriptor, &m->mode, sizeof(mode_t));
send_n_data(file_descriptor, &m->uid, sizeof(uid_t));
send_n_data(file_descriptor, &m->gid, sizeof(gid_t));
send_n_data(file_descriptor, &m->mtime_sec, sizeof(time_t));
send_n_data(file_descriptor, &m->mtime_nsec, sizeof(long));
}
FileMetadata *metadata_receive(int file_descriptor) {
int present;
receive_n_data(file_descriptor, &present, sizeof(int));
if (!present)
return NULL;
FileMetadata *m = malloc(sizeof(FileMetadata));
receive_n_data(file_descriptor, &m->mode, sizeof(mode_t));
receive_n_data(file_descriptor, &m->uid, sizeof(uid_t));
receive_n_data(file_descriptor, &m->gid, sizeof(gid_t));
receive_n_data(file_descriptor, &m->mtime_sec, sizeof(time_t));
receive_n_data(file_descriptor, &m->mtime_nsec, sizeof(long));
return m;
}
void file_restore_metadata(const char *path, FileMetadata *metadata) {
if (metadata == NULL)
return;
chmod(path, metadata->mode & 07777);
int chown_ret = chown(path, metadata->uid, metadata->gid);
(void)chown_ret;
struct timespec times[2];
times[0].tv_sec = 0;
times[0].tv_nsec = UTIME_OMIT;
times[1].tv_sec = metadata->mtime_sec;
times[1].tv_nsec = metadata->mtime_nsec;
utimensat(AT_FDCWD, path, times, 0);
}
src/shared/metadata.h
+15
View File
@@ -0,0 +1,15 @@
#ifndef METADATA_H
#define METADATA_H
#include "file.h"
#include <sys/stat.h>
#define FILE_METADATA_WIRE_SIZE (sizeof(mode_t) + sizeof(uid_t) + sizeof(gid_t) + sizeof(time_t) + sizeof(long))
void metadata_to_buf(char **buf, FileMetadata *m);
FileMetadata *metadata_from_buf(char **buf);
void metadata_send(int file_descriptor, FileMetadata *m);
FileMetadata *metadata_receive(int file_descriptor);
void file_restore_metadata(const char *path, FileMetadata *metadata);
#endif
src/shared/multiprocessing.c
+83
View File
@@ -1,8 +1,17 @@
#include "multiprocessing.h"
#include "chunk.h"
#include "compression.h"
#include "config.h"
#include "data.h"
#include "file.h"
#include "log.h"
#include "metadata.h"
#include "protocol.h"
#include "queue.h"
#include "utils.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <threads.h>
PipelineContextSender *pipeline_context_sender_create(Config *config,
@@ -64,3 +73,77 @@ void pipeline_context_receiver_destroy(PipelineContextReceiver *context) {
cnd_destroy(&context->condition_not_empty);
free(context);
}
static void receive_chunk_enqueue(int file_descriptor,
PipelineContextReceiver *context) {
Data *chunk_data = receive_data(file_descriptor);
Data *data_to_process = chunk_data;
if (context->config->use_compression) {
data_to_process = data_decompress(chunk_data);
data_destroy(chunk_data);
}
Chunk *chunk = chunk_deserialize(data_to_process, context->config->use_metadata);
data_destroy(data_to_process);
for (int i = 0; i < chunk->element_count; i++) {
File *file = chunk->items[i];
chunk->items[i] = NULL;
queue_enqueue_multithreaded(context->queue, file, &context->mutex,
&context->condition_not_empty,
&context->condition_not_full);
}
chunk_destroy(chunk);
}
int receive_thread(void *pipeline_context) {
PipelineContextReceiver *context =
(PipelineContextReceiver *)pipeline_context;
mtx_lock(&context->mutex);
int file_descriptor = context->file_descriptor;
Config *config = context->config;
mtx_unlock(&context->mutex);
Status status = receive_status(file_descriptor);
while (status == STATUS_NEXT || status == STATUS_CHUNK) {
if (status == STATUS_CHUNK) {
receive_chunk_enqueue(file_descriptor, context);
} else {
File *file = file_receive(config, file_descriptor);
queue_enqueue_multithreaded(context->queue, file, &context->mutex,
&context->condition_not_empty,
&context->condition_not_full);
}
status = receive_status(file_descriptor);
}
mtx_lock(&context->mutex);
context->receiver_done = true;
cnd_signal(&context->condition_not_empty);
mtx_unlock(&context->mutex);
return thrd_success;
}
int write_thread(void *pipeline_context) {
PipelineContextReceiver *context =
(PipelineContextReceiver *)pipeline_context;
mtx_lock(&context->mutex);
bool save_to_disk = context->config->save_to_disk;
char *root_directory = str_dup(context->config->receive_root_directory);
mtx_unlock(&context->mutex);
while (true) {
File *file = queue_dequeue_multithreaded(
context->queue, &context->mutex, &context->condition_not_empty,
&context->condition_not_full, &context->receiver_done);
if (file == NULL) {
free(root_directory);
return thrd_success;
}
if (save_to_disk) {
char *disk_path = path_cat(root_directory, file->path);
to_disk(disk_path, file->data->data, file->data->size);
file_restore_metadata(disk_path, file->metadata);
free(disk_path);
}
file_destroy(file);
}
}
src/shared/multiprocessing.h
+3
View File
@@ -4,6 +4,7 @@
#include <threads.h>
#include "config.h"
#include "file.h"
#include "queue.h"
typedef struct {
@@ -38,4 +39,6 @@ PipelineContextReceiver *pipeline_context_receiver_create(Config *config,
Queue *queue_receiver,
int file_descriptor);
void pipeline_context_receiver_destroy(PipelineContextReceiver *context);
int receive_thread(void *pipeline_context);
int write_thread(void *pipeline_context);
#endif
src/shared/protocol.c
+124
View File
@@ -0,0 +1,124 @@
#include "protocol.h"
#include "log.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
static __thread int io_read_fd = -1;
static __thread int io_write_fd = -1;
void io_set_fds(int read_fd, int write_fd) {
io_read_fd = read_fd;
io_write_fd = write_fd;
}
static int io_fd(int dir_fd, int file_descriptor) {
return (dir_fd != -1) ? dir_fd : file_descriptor;
}
void send_n_data(int file_descriptor, void *data, size_t data_size) {
log_message(LOG_LEVEL_DEBUG, " Sending n Data: %zu", data_size);
int fd = io_fd(io_write_fd, file_descriptor);
ssize_t total_bytes_send = 0;
while (total_bytes_send < data_size) {
ssize_t bytes_send =
write(fd, (char *)data + total_bytes_send, data_size - total_bytes_send);
if (bytes_send <= 0) {
perror("Could not send data!");
exit(EXIT_FAILURE);
}
total_bytes_send += bytes_send;
}
log_message(LOG_LEVEL_DEBUG, " Send n Data: %zu", total_bytes_send);
}
void receive_n_data(int file_descriptor, void *data, size_t data_size) {
log_message(LOG_LEVEL_DEBUG, " Receiving n Data: %zu", data_size);
int fd = io_fd(io_read_fd, file_descriptor);
size_t total_bytes_received = 0;
while (total_bytes_received < data_size) {
ssize_t bytes_received =
read(fd, (char *)data + total_bytes_received, data_size - total_bytes_received);
if (bytes_received == -1 || bytes_received == 0) {
perror("Could not receive bytes!");
exit(EXIT_FAILURE);
}
total_bytes_received += bytes_received;
}
log_message(LOG_LEVEL_DEBUG, " Received n Data: %zu", total_bytes_received);
}
static const char *status_to_string(Status status) {
switch (status) {
case STATUS_OK:
return "OK";
case STATUS_ERROR:
return "ERROR";
case STATUS_FINISHED:
return "FINISHED";
case STATUS_NEXT:
return "NEXT";
case STATUS_CHUNK:
return "CHUNK";
default:
return "UNKNOWN";
}
}
void send_str(int file_descriptor, char *data) {
size_t size = strlen(data);
send_n_data(file_descriptor, &size, sizeof(size_t));
send_n_data(file_descriptor, data, size);
log_message(LOG_LEVEL_DEBUG, "Send String: %s", data);
}
char *receive_str(int file_descriptor) {
size_t size;
receive_n_data(file_descriptor, &size, sizeof(size_t));
char *data = (char *)malloc(size + 1);
receive_n_data(file_descriptor, data, size);
data[size] = '\0';
log_message(LOG_LEVEL_DEBUG, "Received String: %s", data);
return data;
}
void send_data(int file_descriptor, Data *data) {
unsigned long long data_size = data->size;
send_n_data(file_descriptor, &data_size, sizeof(unsigned long long));
send_n_data(file_descriptor, data->data, data_size);
log_message(LOG_LEVEL_DEBUG, "Send %lld data", data_size);
}
Data *receive_data(int file_descriptor) {
size_t size = 0;
receive_n_data(file_descriptor, &size, sizeof(unsigned long long));
void *data = malloc(size);
receive_n_data(file_descriptor, data, size);
log_message(LOG_LEVEL_DEBUG, "Received %lld data", size);
return data_create(data, size);
}
void send_int(int file_descriptor, int data) {
send_n_data(file_descriptor, &data, sizeof(int));
log_message(LOG_LEVEL_DEBUG, "Send Int: %d", data);
}
int receive_int(int file_descriptor) {
int data;
receive_n_data(file_descriptor, &data, sizeof(int));
log_message(LOG_LEVEL_DEBUG, "Received Int: %d", data);
return data;
}
void send_status(int file_descriptor, Status status) {
send_n_data(file_descriptor, &status, sizeof(Status));
log_message(LOG_LEVEL_DEBUG, "Send Status: %s", status_to_string(status));
}
Status receive_status(int file_descriptor) {
Status data;
receive_n_data(file_descriptor, &data, sizeof(Status));
log_message(LOG_LEVEL_DEBUG, "Received Status: %s", status_to_string(data));
return data;
}
src/shared/protocol.h
+23
View File
@@ -0,0 +1,23 @@
#ifndef PROTOCOL_H
#define PROTOCOL_H
#include "data.h"
#include <stddef.h>
typedef int Status;
enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK };
void io_set_fds(int read_fd, int write_fd);
void send_n_data(int file_descriptor, void *data, size_t data_size);
void receive_n_data(int file_descriptor, void *data, size_t data_size);
void send_str(int file_descriptor, char *data);
char *receive_str(int file_descriptor);
void send_data(int file_descriptor, Data *data);
Data *receive_data(int file_descriptor);
void send_int(int file_descriptor, int data);
int receive_int(int file_descriptor);
void send_status(int file_descriptor, Status status);
Status receive_status(int file_descriptor);
#endif
src/shared/queue.c
+1 -1
View File
@@ -59,7 +59,7 @@ bool queue_is_full(Queue *queue) {
return queue->size == queue->capacity;
}
void queue_double_capacity(Queue *queue) {
static void queue_double_capacity(Queue *queue) {
if (queue == NULL)
return;
unsigned int new_capacity = queue->capacity * 2;
src/shared/queue.h
-1
View File
@@ -17,7 +17,6 @@ Queue *queue_create(int capacity, void (*destroyer)(void *item));
void queue_destroy(Queue *queue);
bool queue_is_empty(Queue *queue);
bool queue_is_full(Queue *queue);
void queue_double_capacity(Queue *queue);
void queue_enqueue(Queue *queue, void *item);
void queue_enqueue_multithreaded(Queue *queue, void *item, mtx_t *mutex,
cnd_t *condition_not_empty,
src/shared/socket.c
-345
View File
@@ -1,345 +0,0 @@
#include "socket.h"
#include "log.h"
#include <arpa/inet.h>
#include <fcntl.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
static __thread int io_read_fd = -1;
static __thread int io_write_fd = -1;
void io_set_fds(int read_fd, int write_fd) {
io_read_fd = read_fd;
io_write_fd = write_fd;
}
static int io_fd(int dir_fd, int file_descriptor) {
return (dir_fd != -1) ? dir_fd : file_descriptor;
}
typedef struct {
char user[256];
char host[256];
char remote_path[4096];
} RemoteDest;
static int parse_remote_dest(const char *dest, RemoteDest *r) {
const char *colon = strchr(dest, ':');
if (!colon) return -1;
size_t remote_path_len = strlen(colon + 1);
if (remote_path_len >= sizeof(r->remote_path)) return -1;
memcpy(r->remote_path, colon + 1, remote_path_len + 1);
const char *at = memchr(dest, '@', colon - dest);
if (at) {
size_t user_len = at - dest;
if (user_len >= sizeof(r->user)) return -1;
memcpy(r->user, dest, user_len);
r->user[user_len] = '\0';
size_t host_len = colon - at - 1;
if (host_len >= sizeof(r->host)) return -1;
memcpy(r->host, at + 1, host_len);
r->host[host_len] = '\0';
} else {
r->user[0] = '\0';
size_t host_len = colon - dest;
if (host_len >= sizeof(r->host)) return -1;
memcpy(r->host, dest, host_len);
r->host[host_len] = '\0';
}
return 0;
}
Client *client_connect_ssh(char *destination) {
RemoteDest r;
if (parse_remote_dest(destination, &r) != 0) {
fprintf(stderr, "Invalid remote destination: %s\n", destination);
exit(EXIT_FAILURE);
}
int sv[2];
if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) < 0) {
perror("socketpair failed");
exit(EXIT_FAILURE);
}
int exec_pipe[2];
if (pipe(exec_pipe) < 0) {
perror("pipe failed");
exit(EXIT_FAILURE);
}
pid_t pid = fork();
if (pid < 0) {
perror("fork failed");
exit(EXIT_FAILURE);
}
if (pid == 0) {
close(sv[0]);
close(exec_pipe[0]);
fcntl(exec_pipe[1], F_SETFD, FD_CLOEXEC);
if (sv[1] != STDIN_FILENO)
dup2(sv[1], STDIN_FILENO);
if (sv[1] != STDOUT_FILENO)
dup2(sv[1], STDOUT_FILENO);
if (sv[1] > 1) close(sv[1]);
char ssh_user[512];
if (r.user[0] != '\0')
snprintf(ssh_user, sizeof(ssh_user), "%s@%s", r.user, r.host);
else
snprintf(ssh_user, sizeof(ssh_user), "%s", r.host);
execlp("ssh", "ssh", "-o", "Compression=no", "-o",
"ControlMaster=no", ssh_user, "fastsync-server", "--stdio",
(char *)NULL);
perror("exec of ssh failed");
(void)write(exec_pipe[1], "x", 1);
_exit(1);
}
close(sv[1]);
close(exec_pipe[1]);
char exec_status;
ssize_t n = read(exec_pipe[0], &exec_status, 1);
close(exec_pipe[0]);
if (n > 0) {
close(sv[0]);
waitpid(pid, NULL, 0);
fprintf(stderr, "Error: could not launch 'fastsync-server --stdio' on remote\n");
exit(EXIT_FAILURE);
}
Client *client = malloc(sizeof(Client));
client->file_descriptor = sv[0];
client->address.sin_family = AF_UNIX;
client->address_length = 0;
client->ssh_child_pid = pid;
return client;
}
Server *server_create(int port) {
Server *server = (Server *)malloc(sizeof(Server));
if (server == NULL) {
perror("Could not allocate space for Server");
exit(EXIT_FAILURE);
}
int file_descriptor = socket(AF_INET, SOCK_STREAM, 0);
if (file_descriptor < 0) {
perror("Could not create Socket!");
exit(EXIT_FAILURE);
}
server->file_descriptor = file_descriptor;
int opt = 1;
if (setsockopt(server->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &opt,
sizeof(opt))) {
perror("Error setting a socket option!");
close(server->file_descriptor);
free(server);
exit(EXIT_FAILURE);
}
server->address.sin_family = AF_INET;
server->address.sin_addr.s_addr = INADDR_ANY;
server->address.sin_port = htons(port);
server->address_length = sizeof(server->address);
if (bind(server->file_descriptor, (struct sockaddr *)&server->address,
server->address_length) < 0) {
perror("Could not bind server");
close(server->file_descriptor);
free(server);
exit(EXIT_FAILURE);
}
return server;
}
void server_delete(Server *server) {
free(server);
server = NULL;
};
void server_listen(Server *server, void (*handler)(int file_descriptor)) {
log_message(LOG_LEVEL_INFO, "Start Listening on Port: %d",
server->address.sin_port);
if (listen(server->file_descriptor, 3) < 0) {
perror("Could not listen on port!");
exit(EXIT_FAILURE);
}
int file_descriptor =
accept(server->file_descriptor, (struct sockaddr *)&server->address,
&server->address_length);
if (file_descriptor < 0) {
perror("Could not accept the connection");
exit(EXIT_FAILURE);
}
log_message(LOG_LEVEL_INFO, "Received Connection");
handler(file_descriptor);
close(server->file_descriptor);
close(file_descriptor);
}
Client *client_create() {
int file_descriptor = socket(AF_INET, SOCK_STREAM, 0);
if (file_descriptor < 0) {
perror("Could not create Socket!");
exit(EXIT_FAILURE);
};
Client *client = (Client *)malloc(sizeof(Client));
client->file_descriptor = file_descriptor;
client->address.sin_family = AF_INET;
client->address_length = sizeof(client->address);
client->ssh_child_pid = -1;
return client;
}
void client_connect(Client *client, char *host, int port) {
client->address.sin_port = htons(port);
if (inet_pton(AF_INET, host, &client->address.sin_addr) <= 0) {
perror("Could not convert host address!");
exit(EXIT_FAILURE);
}
if (connect(client->file_descriptor, (struct sockaddr *)&client->address,
client->address_length) < 0) {
perror("Could not connect to Server!");
exit(EXIT_FAILURE);
}
}
void client_disconnect(Client *client) {
close(client->file_descriptor);
if (client->ssh_child_pid > 0) {
int status;
waitpid(client->ssh_child_pid, &status, 0);
client->ssh_child_pid = -1;
}
}
void client_delete(Client *client) {
if (client == NULL)
return;
free(client);
}
void send_n_data(int file_descriptor, void *data, size_t data_size) {
log_message(LOG_LEVEL_DEBUG, " Sending n Data: %zu", data_size);
int fd = io_fd(io_write_fd, file_descriptor);
ssize_t total_bytes_send = 0;
while (total_bytes_send < data_size) {
ssize_t bytes_send = write(fd, (char *)data + total_bytes_send,
data_size - total_bytes_send);
if (bytes_send <= 0) {
perror("Could not send data!");
exit(EXIT_FAILURE);
}
total_bytes_send += bytes_send;
}
log_message(LOG_LEVEL_DEBUG, " Send n Data: %zu", total_bytes_send);
}
void receive_n_data(int file_descriptor, void *data, size_t data_size) {
log_message(LOG_LEVEL_DEBUG, " Receiving n Data: %zu", data_size);
int fd = io_fd(io_read_fd, file_descriptor);
size_t total_bytes_received = 0;
while (total_bytes_received < data_size) {
ssize_t bytes_received =
read(fd, (char *)data + total_bytes_received,
data_size - total_bytes_received);
if (bytes_received == -1 || bytes_received == 0) {
perror("Could not receive bytes!");
exit(EXIT_FAILURE);
}
total_bytes_received += bytes_received;
}
log_message(LOG_LEVEL_DEBUG, " Received n Data: %zu", total_bytes_received);
}
void send_str(int file_descriptor, char *data) {
size_t size = strlen(data);
send_n_data(file_descriptor, &size, sizeof(size_t));
send_n_data(file_descriptor, data, size);
log_message(LOG_LEVEL_DEBUG, "Send String: %s", data);
}
char *receive_str(int file_descriptor) {
size_t size;
receive_n_data(file_descriptor, &size, sizeof(size_t));
char *data = (char *)malloc(size + 1);
receive_n_data(file_descriptor, data, size);
data[size] = '\0';
log_message(LOG_LEVEL_DEBUG, "Received String: %s", data);
return data;
}
void send_data(int file_descriptor, void *data, unsigned long long data_size) {
send_n_data(file_descriptor, &data_size, sizeof(unsigned long long));
send_n_data(file_descriptor, data, data_size);
log_message(LOG_LEVEL_DEBUG, "Send %lld data", data_size);
}
Data *receive_data(int file_descriptor) {
size_t size = 0;
receive_n_data(file_descriptor, &size, sizeof(unsigned long long));
void *data = malloc(size);
receive_n_data(file_descriptor, data, size);
log_message(LOG_LEVEL_DEBUG, "Received %lld data", size);
return data_create(data, size);
}
void send_int(int file_descriptor, int data) {
send_n_data(file_descriptor, &data, sizeof(int));
log_message(LOG_LEVEL_DEBUG, "Send Int: %d", data);
}
int receive_int(int file_descriptor) {
int data;
receive_n_data(file_descriptor, &data, sizeof(int));
log_message(LOG_LEVEL_DEBUG, "Received Int: %d", data);
return data;
}
const char *status_to_string(Status status) {
switch (status) {
case STATUS_OK:
return "OK";
case STATUS_ERROR:
return "ERROR";
case STATUS_FINISHED:
return "FINISHED";
case STATUS_NEXT:
return "NEXT";
case STATUS_CHUNK:
return "CHUNK";
default:
return "UNKNOWN";
}
}
void send_status(int file_descriptor, Status status) {
send_n_data(file_descriptor, &status, sizeof(Status));
log_message(LOG_LEVEL_DEBUG, "Send Status: %s", status_to_string(status));
}
Status receive_status(int file_descriptor) {
Status data;
receive_n_data(file_descriptor, &data, sizeof(Status));
log_message(LOG_LEVEL_DEBUG, "Received Status: %s", status_to_string(data));
return data;
}
src/shared/socket.h
-45
View File
@@ -1,45 +0,0 @@
#ifndef SOCKET_H
#define SOCKET_H
#include "data.h"
#include <netinet/in.h>
typedef int Status;
enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK };
typedef struct Server {
struct sockaddr_in address;
unsigned int address_length;
int file_descriptor;
} Server;
Server *server_create(int port);
void server_listen(Server *server, void (*handler)(int file_descriptor));
void server_delete(Server *server);
typedef struct Client {
struct sockaddr_in address;
unsigned int address_length;
int file_descriptor;
pid_t ssh_child_pid;
} Client;
Client *client_create();
void client_disconnect(Client *client);
void client_delete(Client *client);
void client_connect(Client *client, char *host, int port);
Client *client_connect_ssh(char *destination);
void io_set_fds(int read_fd, int write_fd);
void send_n_data(int file_descriptor, void *data, size_t data_size);
void receive_n_data(int file_descriptor, void *data, size_t data_size);
void send_str(int file_descriptor, char *data);
char *receive_str(int file_descriptor);
void send_data(int file_descriptor, void *data, unsigned long long data_size);
Data *receive_data(int file_descriptor);
void send_int(int file_descriptor, int data);
int receive_int(int file_descriptor);
void send_status(int file_descriptor, Status status);
Status receive_status(int file_descriptor);
#endif
src/shared/transport_ssh.c
+115
View File
@@ -0,0 +1,115 @@
#include "transport_ssh.h"
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
typedef struct {
char user[256];
char host[256];
char remote_path[4096];
} RemoteDest;
static int parse_remote_dest(const char *dest, RemoteDest *r) {
const char *colon = strchr(dest, ':');
if (!colon) return -1;
size_t remote_path_len = strlen(colon + 1);
if (remote_path_len >= sizeof(r->remote_path)) return -1;
memcpy(r->remote_path, colon + 1, remote_path_len + 1);
const char *at = memchr(dest, '@', colon - dest);
if (at) {
size_t user_len = at - dest;
if (user_len >= sizeof(r->user)) return -1;
memcpy(r->user, dest, user_len);
r->user[user_len] = '\0';
size_t host_len = colon - at - 1;
if (host_len >= sizeof(r->host)) return -1;
memcpy(r->host, at + 1, host_len);
r->host[host_len] = '\0';
} else {
r->user[0] = '\0';
size_t host_len = colon - dest;
if (host_len >= sizeof(r->host)) return -1;
memcpy(r->host, dest, host_len);
r->host[host_len] = '\0';
}
return 0;
}
Client *client_connect_ssh(char *destination) {
RemoteDest r;
if (parse_remote_dest(destination, &r) != 0) {
fprintf(stderr, "Invalid remote destination: %s\n", destination);
exit(EXIT_FAILURE);
}
int sv[2];
if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) < 0) {
perror("socketpair failed");
exit(EXIT_FAILURE);
}
int exec_pipe[2];
if (pipe(exec_pipe) < 0) {
perror("pipe failed");
exit(EXIT_FAILURE);
}
pid_t pid = fork();
if (pid < 0) {
perror("fork failed");
exit(EXIT_FAILURE);
}
if (pid == 0) {
close(sv[0]);
close(exec_pipe[0]);
fcntl(exec_pipe[1], F_SETFD, FD_CLOEXEC);
if (sv[1] != STDIN_FILENO)
dup2(sv[1], STDIN_FILENO);
if (sv[1] != STDOUT_FILENO)
dup2(sv[1], STDOUT_FILENO);
if (sv[1] > 1) close(sv[1]);
char ssh_user[512];
if (r.user[0] != '\0')
snprintf(ssh_user, sizeof(ssh_user), "%s@%s", r.user, r.host);
else
snprintf(ssh_user, sizeof(ssh_user), "%s", r.host);
execlp("ssh", "ssh", "-o", "Compression=no", "-o",
"ControlMaster=no", ssh_user, "fastsync-server", "--stdio",
(char *)NULL);
perror("exec of ssh failed");
(void)write(exec_pipe[1], "x", 1);
_exit(1);
}
close(sv[1]);
close(exec_pipe[1]);
char exec_status;
ssize_t n = read(exec_pipe[0], &exec_status, 1);
close(exec_pipe[0]);
if (n > 0) {
close(sv[0]);
waitpid(pid, NULL, 0);
fprintf(stderr, "Error: could not launch 'fastsync-server --stdio' on remote\n");
exit(EXIT_FAILURE);
}
Client *client = malloc(sizeof(Client));
client->file_descriptor = sv[0];
client->address.sin_family = AF_UNIX;
client->address_length = 0;
client->ssh_child_pid = pid;
return client;
}
src/shared/transport_ssh.h
+8
View File
@@ -0,0 +1,8 @@
#ifndef TRANSPORT_SSH_H
#define TRANSPORT_SSH_H
#include "transport_tcp.h"
Client *client_connect_ssh(char *destination);
#endif
src/shared/transport_tcp.c
+119
View File
@@ -0,0 +1,119 @@
#include "transport_tcp.h"
#include "log.h"
#include <arpa/inet.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
Server *server_create(int port) {
Server *server = (Server *)malloc(sizeof(Server));
if (server == NULL) {
perror("Could not allocate space for Server");
exit(EXIT_FAILURE);
}
int file_descriptor = socket(AF_INET, SOCK_STREAM, 0);
if (file_descriptor < 0) {
perror("Could not create Socket!");
exit(EXIT_FAILURE);
}
server->file_descriptor = file_descriptor;
int opt = 1;
if (setsockopt(server->file_descriptor, SOL_SOCKET, SO_REUSEADDR, &opt,
sizeof(opt))) {
perror("Error setting a socket option!");
close(server->file_descriptor);
free(server);
exit(EXIT_FAILURE);
}
server->address.sin_family = AF_INET;
server->address.sin_addr.s_addr = INADDR_ANY;
server->address.sin_port = htons(port);
server->address_length = sizeof(server->address);
if (bind(server->file_descriptor, (struct sockaddr *)&server->address,
server->address_length) < 0) {
perror("Could not bind server");
close(server->file_descriptor);
free(server);
exit(EXIT_FAILURE);
}
return server;
}
void server_delete(Server **server) {
if (server == NULL || *server == NULL) return;
free(*server);
*server = NULL;
}
void server_listen(Server *server, void (*handler)(int file_descriptor)) {
log_message(LOG_LEVEL_INFO, "Start Listening on Port: %d",
server->address.sin_port);
if (listen(server->file_descriptor, 3) < 0) {
perror("Could not listen on port!");
exit(EXIT_FAILURE);
}
int file_descriptor =
accept(server->file_descriptor, (struct sockaddr *)&server->address,
&server->address_length);
if (file_descriptor < 0) {
perror("Could not accept the connection");
exit(EXIT_FAILURE);
}
log_message(LOG_LEVEL_INFO, "Received Connection");
handler(file_descriptor);
close(server->file_descriptor);
close(file_descriptor);
}
Client *client_create() {
int file_descriptor = socket(AF_INET, SOCK_STREAM, 0);
if (file_descriptor < 0) {
perror("Could not create Socket!");
exit(EXIT_FAILURE);
}
Client *client = (Client *)malloc(sizeof(Client));
client->file_descriptor = file_descriptor;
client->address.sin_family = AF_INET;
client->address_length = sizeof(client->address);
client->ssh_child_pid = -1;
return client;
}
void client_connect(Client *client, char *host, int port) {
client->address.sin_port = htons(port);
if (inet_pton(AF_INET, host, &client->address.sin_addr) <= 0) {
perror("Could not convert host address!");
exit(EXIT_FAILURE);
}
if (connect(client->file_descriptor, (struct sockaddr *)&client->address,
client->address_length) < 0) {
perror("Could not connect to Server!");
exit(EXIT_FAILURE);
}
}
void client_disconnect(Client *client) {
close(client->file_descriptor);
if (client->ssh_child_pid > 0) {
int status;
waitpid(client->ssh_child_pid, &status, 0);
client->ssh_child_pid = -1;
}
}
void client_delete(Client *client) {
if (client == NULL)
return;
free(client);
}
src/shared/transport_tcp.h
+28
View File
@@ -0,0 +1,28 @@
#ifndef TRANSPORT_TCP_H
#define TRANSPORT_TCP_H
#include <netinet/in.h>
#include <sys/types.h>
typedef struct Server {
struct sockaddr_in address;
unsigned int address_length;
int file_descriptor;
} Server;
typedef struct Client {
struct sockaddr_in address;
unsigned int address_length;
int file_descriptor;
pid_t ssh_child_pid;
} Client;
Server *server_create(int port);
void server_listen(Server *server, void (*handler)(int file_descriptor));
void server_delete(Server **server);
Client *client_create();
void client_connect(Client *client, char *host, int port);
void client_disconnect(Client *client);
void client_delete(Client *client);
#endif
src/shared/utils.c
+2 -33
View File
@@ -1,12 +1,9 @@
#include "utils.h"
#include "libgen.h"
#include "sys/stat.h"
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <sys/stat.h>
void mkdir_r(char *path) {
char *path_duplicate = malloc(strlen(path) + 1);
@@ -39,7 +36,7 @@ void mkdir_r(char *path) {
free(path_current);
}
char *str_dup(char *string) {
char *str_dup(const char *string) {
if (string == NULL)
return NULL;
char *new_string = (char *)malloc(strlen(string) + 1);
@@ -47,34 +44,6 @@ char *str_dup(char *string) {
return new_string;
}
void to_disk(char *path, void *data, unsigned long long data_size) {
char *directory = str_dup(path);
char *dir_to_free = directory;
directory = dirname(directory);
mkdir_r(directory);
FILE *file_pointer = fopen(path, "wb");
if (file_pointer == NULL) {
perror("Could not open File");
exit(EXIT_FAILURE);
}
fwrite(data, 1, data_size, file_pointer);
fclose(file_pointer);
free(dir_to_free);
}
void file_restore_metadata(const char *path, FileMetadata *metadata) {
if (metadata == NULL)
return;
chmod(path, metadata->mode & 07777);
chown(path, metadata->uid, metadata->gid);
struct timespec times[2];
times[0].tv_sec = 0;
times[0].tv_nsec = UTIME_OMIT;
times[1].tv_sec = metadata->mtime_sec;
times[1].tv_nsec = metadata->mtime_nsec;
utimensat(AT_FDCWD, path, times, 0);
}
char *path_cat(char *path1, char *path2) {
if (path1 == NULL || *path1 == '\0')
return str_dup(path2);
src/shared/utils.h
+1 -5
View File
@@ -1,12 +1,8 @@
#ifndef UTILS_H
#define UTILS_H
#include "file.h"
void mkdir_r(char *path);
char *str_dup(char *string);
void to_disk(char *path, void *data, unsigned long long data_size);
void file_restore_metadata(const char *path, FileMetadata *metadata);
char *str_dup(const char *string);
char *path_cat(char *path1, char *path2);
#endif
tests/test_chunk.c
+1 -69
View File
@@ -80,77 +80,9 @@ static void test_chunk_operations() {
EXPECT_EQ_INT(memcmp(deserialized->items[0]->data->data, content1, len1), 0);
EXPECT_EQ_INT(memcmp(deserialized->items[1]->data->data, content2, len2), 0);
chunk_data_delete(serialized);
data_destroy(serialized);
chunk_destroy(deserialized);
// Test chunk_format layout (old format)
Data *formatted = chunk_format(chunk);
EXPECT_NOT_NULL(formatted);
unsigned long long expected_size =
(sizeof(int) + strlen(path1) + sizeof(int) + sizeof(unsigned long long) + len1) +
(sizeof(int) + strlen(path2) + sizeof(int) + sizeof(unsigned long long) + len2);
EXPECT_EQ_INT((int)formatted->size, (int)expected_size);
char *ptr = (char *)formatted->data;
// File 1
int p_len1;
memcpy(&p_len1, ptr, sizeof(int));
ptr += sizeof(int);
EXPECT_EQ_INT(p_len1, (int)strlen(path1));
char read_path1[256];
memcpy(read_path1, ptr, p_len1);
read_path1[p_len1] = '\0';
ptr += p_len1;
EXPECT_EQ_STR(read_path1, path1);
int meta_present1;
memcpy(&meta_present1, ptr, sizeof(int));
ptr += sizeof(int);
EXPECT_EQ_INT(meta_present1, 0);
unsigned long long d_len1;
memcpy(&d_len1, ptr, sizeof(unsigned long long));
ptr += sizeof(unsigned long long);
EXPECT_EQ_INT((int)d_len1, (int)len1);
char read_content1[256];
memcpy(read_content1, ptr, d_len1);
read_content1[d_len1] = '\0';
ptr += d_len1;
EXPECT_EQ_STR(read_content1, content1);
// File 2
int p_len2;
memcpy(&p_len2, ptr, sizeof(int));
ptr += sizeof(int);
EXPECT_EQ_INT(p_len2, (int)strlen(path2));
char read_path2[256];
memcpy(read_path2, ptr, p_len2);
read_path2[p_len2] = '\0';
ptr += p_len2;
EXPECT_EQ_STR(read_path2, path2);
int meta_present2;
memcpy(&meta_present2, ptr, sizeof(int));
ptr += sizeof(int);
EXPECT_EQ_INT(meta_present2, 0);
unsigned long long d_len2;
memcpy(&d_len2, ptr, sizeof(unsigned long long));
ptr += sizeof(unsigned long long);
EXPECT_EQ_INT((int)d_len2, (int)len2);
char read_content2[256];
memcpy(read_content2, ptr, d_len2);
read_content2[d_len2] = '\0';
ptr += d_len2;
EXPECT_EQ_STR(read_content2, content2);
chunk_data_delete(formatted);
chunk_destroy(chunk);
unlink(path1);
tests/test_compression.c
+1
View File
@@ -1,5 +1,6 @@
#include "test_utils.h"
#include "chunk.h"
#include "compression.h"
#include "data.h"
#include "file.h"
#include "utils.h"
tests/test_config.c
+28 -3
View File
@@ -26,12 +26,35 @@ static void test_config_lifecycle() {
static void test_config_ssh_dest() {
Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("user@host:/dst"),
true, false, false, false, false, 1, 4, false);
cfg->transport = TRANSPORT_SSH;
cfg->ssh_destination = str_dup("user@host:/dst");
EXPECT_NOT_NULL(cfg);
EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP);
EXPECT_NULL(cfg->ssh_destination);
EXPECT_EQ_STR(cfg->receive_root_directory, "user@host:/dst");
config_parse_ssh_dest(cfg);
EXPECT_EQ_INT(cfg->transport, TRANSPORT_SSH);
EXPECT_EQ_STR(cfg->ssh_destination, "user@host:/dst");
EXPECT_EQ_STR(cfg->receive_root_directory, "user@host:/dst");
EXPECT_EQ_STR(cfg->receive_root_directory, "/dst");
config_delete(cfg);
}
static void test_config_ssh_dest_local_path() {
Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("/local/path"),
true, false, false, false, false, 1, 4, false);
config_parse_ssh_dest(cfg);
EXPECT_EQ_INT(cfg->transport, TRANSPORT_TCP);
EXPECT_NULL(cfg->ssh_destination);
EXPECT_EQ_STR(cfg->receive_root_directory, "/local/path");
config_delete(cfg);
}
static void test_config_ssh_dest_no_user() {
Config *cfg = config_create(str_dup("1.0"), str_dup("/src"), str_dup("host:/remote"),
true, false, false, false, false, 1, 4, false);
config_parse_ssh_dest(cfg);
EXPECT_EQ_INT(cfg->transport, TRANSPORT_SSH);
EXPECT_EQ_STR(cfg->ssh_destination, "host:/remote");
EXPECT_EQ_STR(cfg->receive_root_directory, "/remote");
config_delete(cfg);
}
@@ -70,6 +93,8 @@ static void test_pipeline_receiver_lifecycle() {
void test_config() {
test_config_lifecycle();
test_config_ssh_dest();
test_config_ssh_dest_local_path();
test_config_ssh_dest_no_user();
test_pipeline_sender_lifecycle();
test_pipeline_receiver_lifecycle();
}