Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 6ada6a5aa7 | |||
| a80cb926d2 | |||
| ffa5f3bf57 | |||
| da6b42a418 | |||
| 3d174de5e7 | |||
| d9ab4775fb | |||
| ef267c94e2 | |||
| 6c30557666 | |||
| 091baffbda | |||
| 96c4a1ac47 |
+16
-2
@@ -88,7 +88,14 @@ int main(int argc, char* argv[]) {
|
||||
} else if (strcmp(argv[i], "-n") == 0 || strcmp(argv[i], "--dry-run") == 0) {
|
||||
config->dry_run = true;
|
||||
} else if (strcmp(argv[i], "-p") == 0 && i + 1 < argc) {
|
||||
config->ssh_port = atoi(argv[++i]);
|
||||
char* end;
|
||||
long p = strtol(argv[++i], &end, 10);
|
||||
if (*end != '\0' || p <= 0 || p > 65535) {
|
||||
fprintf(stderr, "Error: invalid SSH port '%s' (must be 1-65535)\n", argv[i]);
|
||||
exit_code = 1;
|
||||
goto cleanup;
|
||||
}
|
||||
config->ssh_port = (int)p;
|
||||
} else if (strcmp(argv[i], "--delete") == 0) {
|
||||
config->use_delete = true;
|
||||
} else if (strcmp(argv[i], "--exclude") == 0 && i + 1 < argc) {
|
||||
@@ -165,7 +172,14 @@ int main(int argc, char* argv[]) {
|
||||
free(config->server_host);
|
||||
config->server_host = str_dup(argv[++i]);
|
||||
} else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) {
|
||||
config->server_port = atoi(argv[++i]);
|
||||
char* end;
|
||||
long p = strtol(argv[++i], &end, 10);
|
||||
if (*end != '\0' || p <= 0 || p > 65535) {
|
||||
fprintf(stderr, "Error: invalid server port '%s' (must be 1-65535)\n", argv[i]);
|
||||
exit_code = 1;
|
||||
goto cleanup;
|
||||
}
|
||||
config->server_port = (int)p;
|
||||
} else if (strcmp(argv[i], "--bwlimit") == 0 && i + 1 < argc) {
|
||||
char* end;
|
||||
errno = 0;
|
||||
|
||||
+15
-6
@@ -113,13 +113,12 @@ void handler(int file_descriptor) {
|
||||
}
|
||||
|
||||
static Server* g_server = NULL;
|
||||
static volatile sig_atomic_t g_server_cleanup_requested = 0;
|
||||
|
||||
static void cleanup(int sig) {
|
||||
(void)sig;
|
||||
if (g_server) {
|
||||
server_delete(&g_server);
|
||||
}
|
||||
_exit(0);
|
||||
server_request_shutdown();
|
||||
g_server_cleanup_requested = 1;
|
||||
}
|
||||
|
||||
static void print_server_usage(void) {
|
||||
@@ -182,8 +181,12 @@ int main(int argc, char* argv[]) {
|
||||
log_message(LOG_LEVEL_WARNING, "--ca has no effect without --tls");
|
||||
}
|
||||
|
||||
signal(SIGINT, cleanup);
|
||||
signal(SIGTERM, cleanup);
|
||||
struct sigaction sa;
|
||||
sigemptyset(&sa.sa_mask);
|
||||
sa.sa_handler = cleanup;
|
||||
sa.sa_flags = 0; /* Do NOT set SA_RESTART — we need accept() to return EINTR */
|
||||
sigaction(SIGINT, &sa, NULL);
|
||||
sigaction(SIGTERM, &sa, NULL);
|
||||
g_server = server_create(port);
|
||||
if (g_server == NULL) {
|
||||
log_message(LOG_LEVEL_ERROR, "Failed to create server");
|
||||
@@ -205,5 +208,11 @@ int main(int argc, char* argv[]) {
|
||||
} else {
|
||||
server_listen(g_server, handler);
|
||||
}
|
||||
|
||||
/* Graceful shutdown: if a signal requested cleanup, delete the server */
|
||||
if (g_server_cleanup_requested) {
|
||||
log_message(LOG_LEVEL_INFO, "Shutdown requested, cleaning up");
|
||||
server_delete(&g_server);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -8,6 +8,18 @@
|
||||
|
||||
Data* data_compress(Data* data_to_compress, int compression_level) {
|
||||
log_message(LOG_LEVEL_DEBUG, "Starting to compress data");
|
||||
|
||||
/* Clamp compression level to valid zstd range [1, 22] */
|
||||
if (compression_level < 1) {
|
||||
log_message(LOG_LEVEL_WARNING, "compression_level %d out of range [1,22], using 1",
|
||||
compression_level);
|
||||
compression_level = 1;
|
||||
} else if (compression_level > 22) {
|
||||
log_message(LOG_LEVEL_WARNING, "compression_level %d out of range [1,22], using 22",
|
||||
compression_level);
|
||||
compression_level = 22;
|
||||
}
|
||||
|
||||
size_t dst_size = ZSTD_compressBound(data_to_compress->size);
|
||||
Data* compressed_data = data_create_empty(dst_size);
|
||||
if (compressed_data == NULL)
|
||||
|
||||
+43
-42
@@ -155,6 +155,43 @@ static void* old_data_from_path(const char* full_path, unsigned long long old_si
|
||||
return data;
|
||||
}
|
||||
|
||||
/**
|
||||
* Helper: receive data from wire, optionally decompress, and store in file.
|
||||
* On success, returns the received Data* (caller owns it). On failure, returns NULL.
|
||||
* If `file_data` is received via receive_data(fd), this function handles decompression
|
||||
* when config->use_compression is set.
|
||||
*/
|
||||
static Data* receive_and_decompress(int fd, const Config* config) {
|
||||
Data* file_data = receive_data(fd);
|
||||
if (file_data == NULL)
|
||||
return NULL;
|
||||
if (config->use_compression) {
|
||||
Data* uncompressed = data_decompress(file_data);
|
||||
data_destroy(file_data);
|
||||
if (uncompressed == NULL)
|
||||
return NULL;
|
||||
file_data = uncompressed;
|
||||
}
|
||||
return file_data;
|
||||
}
|
||||
|
||||
/**
|
||||
* Helper: receive metadata from wire and assign to file.
|
||||
* Returns true on success (metadata may be NULL if absent), false on I/O error.
|
||||
*/
|
||||
static bool receive_and_assign_metadata(int fd, const Config* config, File* file) {
|
||||
if (!config->use_metadata)
|
||||
return true;
|
||||
int meta_ok = 1;
|
||||
file->metadata = metadata_receive(fd, &meta_ok);
|
||||
if (!meta_ok) {
|
||||
file_destroy(file);
|
||||
send_status(fd, STATUS_ERROR);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
static File* receive_delta_file(int fd, const Config* config, const char* check_path,
|
||||
void* old_data, unsigned long long old_size) {
|
||||
if (!old_data)
|
||||
@@ -270,34 +307,16 @@ static File* receive_delta_file(int fd, const Config* config, const char* check_
|
||||
return NULL;
|
||||
}
|
||||
|
||||
if (config->use_metadata) {
|
||||
int meta_ok = 1;
|
||||
file->metadata = metadata_receive(fd, &meta_ok);
|
||||
if (!meta_ok) {
|
||||
file_destroy(file);
|
||||
send_status(fd, STATUS_ERROR);
|
||||
return NULL;
|
||||
}
|
||||
}
|
||||
if (!receive_and_assign_metadata(fd, config, file))
|
||||
return NULL;
|
||||
|
||||
Data* file_data = receive_data(fd);
|
||||
Data* file_data = receive_and_decompress(fd, config);
|
||||
if (file_data == NULL) {
|
||||
file_destroy(file);
|
||||
send_status(fd, STATUS_ERROR);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
if (config->use_compression) {
|
||||
Data* uncompressed = data_decompress(file_data);
|
||||
data_destroy(file_data);
|
||||
if (uncompressed == NULL) {
|
||||
file_destroy(file);
|
||||
send_status(fd, STATUS_ERROR);
|
||||
return NULL;
|
||||
}
|
||||
file_data = uncompressed;
|
||||
}
|
||||
|
||||
data_destroy(file->data);
|
||||
file->data = file_data;
|
||||
return file;
|
||||
@@ -375,34 +394,16 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
if (config->use_metadata) {
|
||||
int meta_ok = 1;
|
||||
file->metadata = metadata_receive(fd, &meta_ok);
|
||||
if (!meta_ok) {
|
||||
file_destroy(file);
|
||||
send_status(fd, STATUS_ERROR);
|
||||
return NULL;
|
||||
}
|
||||
}
|
||||
if (!receive_and_assign_metadata(fd, config, file))
|
||||
return NULL;
|
||||
|
||||
Data* file_data = receive_data(fd);
|
||||
Data* file_data = receive_and_decompress(fd, config);
|
||||
if (file_data == NULL) {
|
||||
file_destroy(file);
|
||||
send_status(fd, STATUS_ERROR);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
if (config->use_compression) {
|
||||
Data* uncompressed = data_decompress(file_data);
|
||||
data_destroy(file_data);
|
||||
if (uncompressed == NULL) {
|
||||
file_destroy(file);
|
||||
send_status(fd, STATUS_ERROR);
|
||||
return NULL;
|
||||
}
|
||||
file_data = uncompressed;
|
||||
}
|
||||
|
||||
data_destroy(file->data);
|
||||
file->data = file_data;
|
||||
return file;
|
||||
|
||||
+1
-1
@@ -10,7 +10,7 @@ void set_log_level(LogLevel level) {
|
||||
current_log_level = level;
|
||||
}
|
||||
|
||||
void log_message(LogLevel log_level, char* format, ...) {
|
||||
void log_message(LogLevel log_level, const char* format, ...) {
|
||||
if (log_level < current_log_level)
|
||||
return;
|
||||
time_t now = time(NULL);
|
||||
|
||||
+1
-1
@@ -3,7 +3,7 @@
|
||||
|
||||
typedef enum { LOG_LEVEL_DEBUG, LOG_LEVEL_INFO, LOG_LEVEL_WARNING, LOG_LEVEL_ERROR } LogLevel;
|
||||
|
||||
void log_message(LogLevel log_level, char* message, ...);
|
||||
void log_message(LogLevel log_level, const char* message, ...);
|
||||
void set_log_level(LogLevel level);
|
||||
|
||||
#endif
|
||||
|
||||
+27
-4
@@ -1,6 +1,7 @@
|
||||
#include "protocol.h"
|
||||
#include "log.h"
|
||||
#include <errno.h>
|
||||
#include <openssl/err.h>
|
||||
#include <openssl/ssl.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
@@ -34,11 +35,21 @@ static void bw_throttle(size_t bytes_written) {
|
||||
struct timespec now;
|
||||
clock_gettime(CLOCK_MONOTONIC, &now);
|
||||
|
||||
long long elapsed_ns =
|
||||
(now.tv_sec - bw_last_refill.tv_sec) * 1000000000LL + (now.tv_nsec - bw_last_refill.tv_nsec);
|
||||
/* Use unsigned long long for elapsed_ns to avoid overflow in multiplication.
|
||||
* time_t differences fit comfortably in 64-bit for any practical runtime.
|
||||
* Handle nanosecond carry/borrow to avoid unsigned wraparound when
|
||||
* now.tv_nsec < bw_last_refill.tv_nsec. */
|
||||
long long sec_diff = (long long)(now.tv_sec - bw_last_refill.tv_sec);
|
||||
long long nsec_diff = (long long)(now.tv_nsec - bw_last_refill.tv_nsec);
|
||||
if (nsec_diff < 0) {
|
||||
sec_diff--;
|
||||
nsec_diff += 1000000000LL;
|
||||
}
|
||||
unsigned long long elapsed_ns =
|
||||
(unsigned long long)sec_diff * 1000000000ULL + (unsigned long long)nsec_diff;
|
||||
bw_last_refill = now;
|
||||
|
||||
long long tokens_to_add = (long long)((double)io_bwlimit * elapsed_ns / 1000000000.0);
|
||||
long long tokens_to_add = (long long)((double)io_bwlimit * (double)elapsed_ns / 1000000000.0);
|
||||
bw_tokens += tokens_to_add;
|
||||
if (bw_tokens > (long long)io_bwlimit)
|
||||
bw_tokens = (long long)io_bwlimit;
|
||||
@@ -46,7 +57,9 @@ static void bw_throttle(size_t bytes_written) {
|
||||
bw_tokens -= (long long)bytes_written;
|
||||
|
||||
if (bw_tokens < 0) {
|
||||
long long deficit_ns = (long long)((double)(-bw_tokens) / io_bwlimit * 1000000000.0);
|
||||
long long deficit_ns = (long long)((double)(-bw_tokens) / (double)io_bwlimit * 1000000000.0);
|
||||
if (deficit_ns < 0)
|
||||
deficit_ns = 0;
|
||||
struct timespec sleep_time, remaining;
|
||||
sleep_time.tv_sec = deficit_ns / 1000000000LL;
|
||||
sleep_time.tv_nsec = deficit_ns % 1000000000LL;
|
||||
@@ -79,6 +92,11 @@ bool send_n_data(int file_descriptor, const void* data, size_t data_size) {
|
||||
else
|
||||
bytes_send = write(fd, (const char*)data + total_bytes_send, chunk);
|
||||
if (bytes_send <= 0) {
|
||||
if (io_ssl) {
|
||||
int ssl_err = SSL_get_error(io_ssl, (int)bytes_send);
|
||||
if (ssl_err == SSL_ERROR_WANT_WRITE || ssl_err == SSL_ERROR_WANT_READ)
|
||||
continue;
|
||||
}
|
||||
log_message(LOG_LEVEL_ERROR, "Could not send data");
|
||||
return false;
|
||||
}
|
||||
@@ -102,6 +120,11 @@ bool receive_n_data(int file_descriptor, void* data, size_t data_size) {
|
||||
bytes_received =
|
||||
read(fd, (char*)data + total_bytes_received, data_size - total_bytes_received);
|
||||
if (bytes_received <= 0) {
|
||||
if (io_ssl) {
|
||||
int ssl_err = SSL_get_error(io_ssl, (int)bytes_received);
|
||||
if (ssl_err == SSL_ERROR_WANT_READ || ssl_err == SSL_ERROR_WANT_WRITE)
|
||||
continue;
|
||||
}
|
||||
if (bytes_received == 0)
|
||||
log_message(LOG_LEVEL_ERROR, "Connection closed while receiving data");
|
||||
else
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
#include "log.h"
|
||||
#include "protocol.h"
|
||||
#include <arpa/inet.h>
|
||||
#include <errno.h>
|
||||
#include <openssl/ssl.h>
|
||||
#include <signal.h>
|
||||
#include <stdio.h>
|
||||
@@ -11,6 +12,15 @@
|
||||
#include <sys/wait.h>
|
||||
#include <unistd.h>
|
||||
|
||||
/* Flag set by server_request_shutdown() to request graceful shutdown
|
||||
of the accept loop. Accessed only from transport_tcp.c so it won't
|
||||
cause linker errors when this file is compiled into client/test targets. */
|
||||
static volatile sig_atomic_t g_tcp_cleanup_requested = 0;
|
||||
|
||||
void server_request_shutdown(void) {
|
||||
g_tcp_cleanup_requested = 1;
|
||||
}
|
||||
|
||||
Server* server_create(int port) {
|
||||
Server* server = (Server*)malloc(sizeof(Server));
|
||||
if (server == NULL) {
|
||||
@@ -69,11 +79,13 @@ static void accept_loop(Server* server, void (*child_fn)(int, void*), void* chil
|
||||
return;
|
||||
}
|
||||
signal(SIGCHLD, SIG_IGN);
|
||||
while (1) {
|
||||
while (!g_tcp_cleanup_requested) {
|
||||
struct sockaddr_in client_addr;
|
||||
socklen_t client_len = sizeof(client_addr);
|
||||
int fd = accept(server->file_descriptor, (struct sockaddr*)&client_addr, &client_len);
|
||||
if (fd < 0) {
|
||||
if (errno == EINTR)
|
||||
break;
|
||||
perror("Could not accept the connection");
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -26,6 +26,7 @@ bool server_listen(Server* server, void (*handler)(int file_descriptor));
|
||||
void server_accept_loop(Server* server, void (*child_fn)(int, void*), void* child_ctx,
|
||||
const char* log_fmt);
|
||||
void server_delete(Server** server);
|
||||
void server_request_shutdown(void);
|
||||
Client* client_create();
|
||||
bool client_connect(Client* client, char* host, int port);
|
||||
void client_disconnect(Client* client);
|
||||
|
||||
+30
-1
@@ -61,6 +61,23 @@ char* str_dup(const char* string) {
|
||||
bool glob_match(const char* pattern, const char* str) {
|
||||
while (*pattern) {
|
||||
if (*pattern == '*') {
|
||||
/* Check for double-star (globstar) pattern */
|
||||
if (*(pattern + 1) == '*') {
|
||||
pattern += 2;
|
||||
/* Trailing double-star matches everything */
|
||||
if (*pattern == '\0')
|
||||
return true;
|
||||
/* double-star slash: match at any depth */
|
||||
if (*pattern == '/')
|
||||
pattern++;
|
||||
while (*str) {
|
||||
if (glob_match(pattern, str))
|
||||
return true;
|
||||
str++;
|
||||
}
|
||||
return glob_match(pattern, str);
|
||||
}
|
||||
/* Single * — does not cross / boundaries */
|
||||
pattern++;
|
||||
while (*str && *str != '/') {
|
||||
if (glob_match(pattern, str))
|
||||
@@ -74,8 +91,20 @@ bool glob_match(const char* pattern, const char* str) {
|
||||
pattern++;
|
||||
str++;
|
||||
} else {
|
||||
if (*pattern != *str)
|
||||
if (*pattern != *str) {
|
||||
/* If pattern has a '/' followed by '**', allow zero path components.
|
||||
* Only skip slash-double-star if the remainder is empty or starts with '/';
|
||||
* otherwise fall through to normal character matching. */
|
||||
if (*pattern == '/' && *(pattern + 1) == '*' && *(pattern + 2) == '*') {
|
||||
const char* rest = pattern + 3;
|
||||
if (*rest == '\0')
|
||||
return glob_match(rest, str);
|
||||
if (*rest == '/')
|
||||
return glob_match(rest + 1, str);
|
||||
/* slash-double-star X where X does not start with '/' — fall through */
|
||||
}
|
||||
return false;
|
||||
}
|
||||
pattern++;
|
||||
str++;
|
||||
}
|
||||
|
||||
+1
-1
@@ -271,7 +271,7 @@ void test_file() {
|
||||
test_to_disk_basic();
|
||||
test_to_disk_creates_dirs();
|
||||
test_file_content_to_buffer();
|
||||
if (!getenv("FASTSYNC_UNDER_VALGRIND")) {
|
||||
if (!is_running_under_valgrind()) {
|
||||
// Fork tests are skipped under valgrind because the parent process runs
|
||||
// orders of magnitude slower than the child (parent is instrumented, child
|
||||
// is not), which causes pipe-based protocol handshake timeouts. The parent
|
||||
|
||||
@@ -49,6 +49,33 @@ static void test_glob_question_star() {
|
||||
EXPECT_TRUE(glob_match("?*.txt", "a.txt"));
|
||||
}
|
||||
|
||||
static void test_glob_doublestar_match_all() {
|
||||
EXPECT_TRUE(glob_match("**", "anything"));
|
||||
EXPECT_TRUE(glob_match("**", "path/to/file"));
|
||||
}
|
||||
|
||||
static void test_glob_doublestar_prefix() {
|
||||
EXPECT_TRUE(glob_match("**/foo", "foo"));
|
||||
EXPECT_TRUE(glob_match("**/foo", "bar/foo"));
|
||||
EXPECT_TRUE(glob_match("**/foo", "a/b/c/foo"));
|
||||
EXPECT_FALSE(glob_match("**/foo", "foobar"));
|
||||
EXPECT_FALSE(glob_match("**/foo", "bar/foobar"));
|
||||
}
|
||||
|
||||
static void test_glob_doublestar_suffix() {
|
||||
EXPECT_TRUE(glob_match("foo/**", "foo"));
|
||||
EXPECT_TRUE(glob_match("foo/**", "foo/bar"));
|
||||
EXPECT_TRUE(glob_match("foo/**", "foo/bar/baz"));
|
||||
EXPECT_FALSE(glob_match("foo/**", "foobar"));
|
||||
}
|
||||
|
||||
static void test_glob_doublestar_mid() {
|
||||
EXPECT_TRUE(glob_match("a/**/b", "a/b"));
|
||||
EXPECT_TRUE(glob_match("a/**/b", "a/x/b"));
|
||||
EXPECT_TRUE(glob_match("a/**/b", "a/x/y/z/b"));
|
||||
EXPECT_FALSE(glob_match("a/**/b", "a/x/bad"));
|
||||
}
|
||||
|
||||
void test_glob() {
|
||||
test_glob_exact_match();
|
||||
test_glob_question_mark();
|
||||
@@ -60,4 +87,8 @@ void test_glob() {
|
||||
test_glob_slash_not_matched();
|
||||
test_glob_complex();
|
||||
test_glob_question_star();
|
||||
test_glob_doublestar_match_all();
|
||||
test_glob_doublestar_prefix();
|
||||
test_glob_doublestar_suffix();
|
||||
test_glob_doublestar_mid();
|
||||
}
|
||||
|
||||
@@ -2,9 +2,24 @@
|
||||
#define TEST_UTILS_H
|
||||
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <stdbool.h>
|
||||
|
||||
// Detect if running under valgrind by checking /proc/self/maps for vgpreload.
|
||||
// This is used to skip fork-based tests that are incompatible with valgrind
|
||||
// (the instrumented parent runs too slowly, causing pipe timeouts).
|
||||
static inline bool is_running_under_valgrind(void) {
|
||||
FILE* f = fopen("/proc/self/maps", "r");
|
||||
if (!f)
|
||||
return false;
|
||||
char buf[4096];
|
||||
size_t n = fread(buf, 1, sizeof(buf) - 1, f);
|
||||
fclose(f);
|
||||
buf[n] = '\0';
|
||||
return strstr(buf, "vgpreload") != NULL;
|
||||
}
|
||||
|
||||
// Global test suite status
|
||||
extern int tests_run;
|
||||
extern int tests_failed;
|
||||
|
||||
Reference in New Issue
Block a user