Multi-client server, protocol versioning, streaming decompression #11

Merged
TapTap merged 2 commits from server-improvements into main 2026-07-16 15:15:20 +02:00
7 changed files with 79 additions and 20 deletions
+1 -1
View File
@@ -56,7 +56,7 @@ int main(int argc, char *argv[]) {
save_to_disk = true; save_to_disk = true;
} }
Config *config = config_create(str_dup("1.0.0"), NULL, NULL, Config *config = config_create(str_dup(PROTOCOL_VERSION), NULL, NULL,
save_to_disk, false, false, false, false, 5, false, 0); save_to_disk, false, false, false, false, 5, false, 0);
int positional_args[2]; int positional_args[2];
+15 -4
View File
@@ -142,6 +142,16 @@ void handler(int file_descriptor) {
close(file_descriptor); close(file_descriptor);
} }
static Server *g_server = NULL;
static void cleanup(int sig) {
(void)sig;
if (g_server) {
server_delete(&g_server);
}
_exit(0);
}
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
signal(SIGPIPE, SIG_IGN); signal(SIGPIPE, SIG_IGN);
for (int i = 1; i < argc; i++) { for (int i = 1; i < argc; i++) {
@@ -153,12 +163,13 @@ int main(int argc, char *argv[]) {
set_log_level(LOG_LEVEL_DEBUG); set_log_level(LOG_LEVEL_DEBUG);
} }
} }
Server *server = server_create(8080); signal(SIGINT, cleanup);
if (server == NULL) { signal(SIGTERM, cleanup);
g_server = server_create(8080);
if (g_server == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to create server"); log_message(LOG_LEVEL_ERROR, "Failed to create server");
return 1; return 1;
} }
server_listen(server, handler); server_listen(g_server, handler);
server_delete(&server);
return 0; return 0;
} }
+26 -5
View File
@@ -4,6 +4,8 @@
#include "stdlib.h" #include "stdlib.h"
#include "zstd.h" #include "zstd.h"
#define INITIAL_DECOMPRESS_BUF_SIZE (1024 * 1024)
Data *data_compress(Data *data_to_compress, int compression_level) { Data *data_compress(Data *data_to_compress, int compression_level) {
log_message(LOG_LEVEL_DEBUG, "Starting to compress data"); log_message(LOG_LEVEL_DEBUG, "Starting to compress data");
size_t dst_size = ZSTD_compressBound(data_to_compress->size); size_t dst_size = ZSTD_compressBound(data_to_compress->size);
@@ -50,19 +52,25 @@ Data *data_decompress(Data *compressed_data) {
return NULL; return NULL;
} }
Data *uncompressed_data = data_create_empty((size_t)dst_size);
if (uncompressed_data == NULL) return NULL;
ZSTD_DCtx *dctx = ZSTD_createDCtx(); ZSTD_DCtx *dctx = ZSTD_createDCtx();
if (!dctx) { if (!dctx) {
log_message(LOG_LEVEL_ERROR, log_message(LOG_LEVEL_ERROR,
"Failed to create ZSTD decompression context"); "Failed to create ZSTD decompression context");
data_destroy(uncompressed_data); return NULL;
}
size_t buf_size = (!ZSTD_isError(dst_size) && dst_size > 0)
? (size_t)dst_size
: INITIAL_DECOMPRESS_BUF_SIZE;
Data *uncompressed_data = data_create_empty(buf_size);
if (!uncompressed_data) {
log_message(LOG_LEVEL_ERROR, "Failed to allocate decompression buffer");
ZSTD_freeDCtx(dctx);
return NULL; return NULL;
} }
ZSTD_inBuffer input = {compressed_data->data, compressed_data->size, 0}; ZSTD_inBuffer input = {compressed_data->data, compressed_data->size, 0};
ZSTD_outBuffer output = {uncompressed_data->data, (size_t)dst_size, 0}; ZSTD_outBuffer output = {uncompressed_data->data, buf_size, 0};
size_t ret; size_t ret;
do { do {
@@ -74,6 +82,19 @@ Data *data_decompress(Data *compressed_data) {
data_destroy(uncompressed_data); data_destroy(uncompressed_data);
return NULL; return NULL;
} }
if (ret > 0 && output.pos == output.size) {
buf_size *= 2;
void *new_data = realloc(uncompressed_data->data, buf_size);
if (!new_data) {
log_message(LOG_LEVEL_ERROR, "Failed to grow decompression buffer");
ZSTD_freeDCtx(dctx);
data_destroy(uncompressed_data);
return NULL;
}
uncompressed_data->data = new_data;
output.dst = new_data;
output.size = buf_size;
}
} while (ret > 0); } while (ret > 0);
uncompressed_data->size = output.pos; uncompressed_data->size = output.pos;
+8
View File
@@ -103,6 +103,14 @@ Config *config_receive(int file_descriptor) {
if (config == NULL) return NULL; if (config == NULL) return NULL;
config->version = receive_str(file_descriptor); config->version = receive_str(file_descriptor);
if (!config->version) { free(config); return NULL; } if (!config->version) { free(config); return NULL; }
if (strcmp(config->version, PROTOCOL_VERSION) != 0) {
fprintf(stderr, "Protocol version mismatch: client=%s, server=%s\n",
config->version, PROTOCOL_VERSION);
free(config->version);
free(config);
send_status(file_descriptor, STATUS_ERROR);
return NULL;
}
config->send_directory = receive_str(file_descriptor); config->send_directory = receive_str(file_descriptor);
if (!config->send_directory) { free(config->version); free(config); return NULL; } if (!config->send_directory) { free(config->version); free(config); return NULL; }
config->receive_root_directory = receive_str(file_descriptor); config->receive_root_directory = receive_str(file_descriptor);
+1
View File
@@ -34,6 +34,7 @@ typedef struct Config {
unsigned long long min_size; unsigned long long min_size;
} Config; } Config;
#define PROTOCOL_VERSION "1.0.0"
#define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024) #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
Config *config_create(char *version, char *send_directory, Config *config_create(char *version, char *send_directory,
+5
View File
@@ -10,6 +10,7 @@
#include <unistd.h> #include <unistd.h>
#include "compression.h" #include "compression.h"
#include "log.h"
#include "config.h" #include "config.h"
#include "data.h" #include "data.h"
#include "file.h" #include "file.h"
@@ -103,6 +104,10 @@ bool file_send_single_calls(File *file, int file_descriptor, bool use_metadata,
return false; return false;
} }
data_destroy(file->data); data_destroy(file->data);
if (compressed_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Compression failed in file_send_single_calls");
exit(EXIT_FAILURE);
}
file->data = compressed_data; file->data = compressed_data;
} }
if (!send_str(file_descriptor, file->path)) return false; if (!send_str(file_descriptor, file->path)) return false;
+17 -4
View File
@@ -1,6 +1,7 @@
#include "transport_tcp.h" #include "transport_tcp.h"
#include "log.h" #include "log.h"
#include <arpa/inet.h> #include <arpa/inet.h>
#include <signal.h>
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
@@ -49,6 +50,7 @@ Server *server_create(int port) {
void server_delete(Server **server) { void server_delete(Server **server) {
if (server == NULL || *server == NULL) return; if (server == NULL || *server == NULL) return;
close((*server)->file_descriptor);
free(*server); free(*server);
*server = NULL; *server = NULL;
} }
@@ -61,17 +63,28 @@ bool server_listen(Server *server, void (*handler)(int file_descriptor)) {
return false; return false;
} }
signal(SIGCHLD, SIG_IGN);
while (1) {
struct sockaddr_in client_addr;
socklen_t client_len = sizeof(client_addr);
int file_descriptor = int file_descriptor =
accept(server->file_descriptor, (struct sockaddr *)&server->address, accept(server->file_descriptor, (struct sockaddr *)&client_addr,
&server->address_length); &client_len);
if (file_descriptor < 0) { if (file_descriptor < 0) {
perror("Could not accept the connection"); perror("Could not accept the connection");
return false; continue;
} }
log_message(LOG_LEVEL_INFO, "Received Connection"); log_message(LOG_LEVEL_INFO, "Received Connection");
handler(file_descriptor); pid_t pid = fork();
if (pid == 0) {
close(server->file_descriptor); close(server->file_descriptor);
handler(file_descriptor);
close(file_descriptor); close(file_descriptor);
_exit(0);
}
close(file_descriptor);
}
return true; return true;
} }