refactor: split monolithic modules and extract shared components
- Remove dead socket.c/socket.h (accumulated duplicate symbols) - Extract compression.h/c (data_compress/decompress from data.c) - Extract io.h/c (low-level I/O from protocol.c) - Extract metadata.h/c (unified metadata wire format from file.c, chunk.c, utils.c) - Split client.c into client_cli.c (CLI parsing) + client_send.c (send logic) - Move receiver pipeline threads from server.c to multiprocessing.c - Move to_disk/file_restore_metadata from utils.c to file.c/metadata.c - Fix const qualifiers on to_disk and str_dup signatures - Suppress chown unused-result warning
This commit is contained in:
@@ -0,0 +1,188 @@
|
||||
#include "client_send.h"
|
||||
#include "chunk.h"
|
||||
#include "compression.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 {
|
||||
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;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
Reference in New Issue
Block a user