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

Merge remote-tracking branch 'origin/fix/performance' into merge-all-v2

Browse Source 
# Conflicts:
#	src/client/client_send.c
#	src/client/scanner.h
#	src/shared/compression.c
This commit is contained in:
TapTap
2026-07-21 17:53:02 +02:00
parent 909d84b36e 4bba1e8c02
commit 2e73c6dc13
7 changed files with 391 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/client_send.c
+76 -185
View File
@@ -16,22 +16,14 @@
#include "transport_ssh.h" #include "transport_ssh.h"
#include "transport_tls.h" #include "transport_tls.h"
#include "utils.h" #include "utils.h"
#include <signal.h>
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <threads.h> #include <threads.h>
#include <time.h> #include <time.h>
#include <unistd.h>
static volatile sig_atomic_t g_abort_requested = 0; #define STREAM_THRESHOLD (64ULL * 1024 * 1024)
static int g_abort_fd = -1;
static void handle_sigint(int sig) {
(void)sig;
g_abort_requested = 1;
}
#define KEEPALIVE_INTERVAL 30
static int incremental_check(Client* client, File* file, DeltaSignature** out_sig) { static int incremental_check(Client* client, File* file, DeltaSignature** out_sig) {
*out_sig = NULL; *out_sig = NULL;
@@ -107,35 +99,6 @@ static int send_delta(Client* client, File* file, DeltaSignature* sig, Config* c
return ok ? 0 : -1; return ok ? 0 : -1;
} }
static bool batch_incremental_check(Client* client, ArrayList* files) {
if (!send_status(client->file_descriptor, STATUS_CHECK_BATCH))
return false;
if (!send_int(client->file_descriptor, files->size))
return false;
for (int i = 0; i < files->size; i++) {
File* file = (File*)files->items[i];
if (!send_str(client->file_descriptor, file->path))
return false;
unsigned long long fsize = file->data ? file->data->size : 0;
long long mtime = file->metadata ? file->metadata->mtime_sec : 0;
if (!send_n_data(client->file_descriptor, &fsize, sizeof(fsize)))
return false;
if (!send_n_data(client->file_descriptor, &mtime, sizeof(mtime)))
return false;
}
for (int i = 0; i < files->size; i++) {
Status s;
if (!receive_status(client->file_descriptor, &s))
return false;
File* file = (File*)files->items[i];
if (s == STATUS_OK)
file->skip = true;
else if (s == STATUS_ERROR)
return false;
}
return true;
}
typedef bool (*file_send_fn)(File*, int, bool, int, bool); typedef bool (*file_send_fn)(File*, int, bool, int, bool);
// Send a single file directly (non-incremental path). // Send a single file directly (non-incremental path).
@@ -158,9 +121,6 @@ static int send_single_file(Client* client, File* file, Config* config, bool use
bool use_sendfile) { bool use_sendfile) {
int compression_level = config->use_compression ? config->compression_level : 0; int compression_level = config->use_compression ? config->compression_level : 0;
if (file->skip)
return 1;
if (!use_incremental) { if (!use_incremental) {
if (use_sendfile) { if (use_sendfile) {
return send_file_direct_sendfile(file, client->file_descriptor, config->use_metadata) ? 0 return send_file_direct_sendfile(file, client->file_descriptor, config->use_metadata) ? 0
@@ -251,10 +211,13 @@ int send_chunk(Client* client, Chunk* chunk, Config* config) {
return 0; return 0;
} }
bool use_sendfile = config->use_sendfile && !config->use_compression;
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
int rc = File* f = chunk->items[i];
send_single_file(client, chunk->items[i], config, config->use_incremental, use_sendfile); if (f == NULL)
continue;
bool stream = f->data->data == NULL && f->data->size > 0;
bool use_sendfile = (config->use_sendfile && !config->use_compression) || stream;
int rc = send_single_file(client, f, config, config->use_incremental, use_sendfile);
if (rc == 1) if (rc == 1)
continue; continue;
if (rc < 0) if (rc < 0)
@@ -298,30 +261,7 @@ static int send_chunks_multithreaded(void* pipeline_context) {
return thrd_error; return thrd_error;
} }
g_abort_fd = client->file_descriptor;
time_t last_activity = time(NULL);
while (true) { while (true) {
if (g_abort_requested) {
send_status(client->file_descriptor, STATUS_ABORT);
client_disconnect(client);
client_delete(client);
return thrd_error;
}
time_t now = time(NULL);
if (now - last_activity >= KEEPALIVE_INTERVAL) {
if (!send_status(client->file_descriptor, STATUS_KEEPALIVE)) {
client_disconnect(client);
client_delete(client);
return thrd_error;
}
Status s;
if (!receive_status(client->file_descriptor, &s)) {
client_disconnect(client);
client_delete(client);
return thrd_error;
}
last_activity = now;
}
Chunk* current_chunk = queue_dequeue_multithreaded( Chunk* current_chunk = queue_dequeue_multithreaded(
context->queue_loader, &context->mutex_loader, &context->condition_not_empty_loader, context->queue_loader, &context->mutex_loader, &context->condition_not_empty_loader,
&context->condition_not_full_loader, &context->loader_done); &context->condition_not_full_loader, &context->loader_done);
@@ -361,16 +301,14 @@ static int send_chunks_multithreaded(void* pipeline_context) {
static int scan_directory_multithreaded(void* pipeline_context) { static int scan_directory_multithreaded(void* pipeline_context) {
PipelineContextSender* context = (PipelineContextSender*)pipeline_context; PipelineContextSender* context = (PipelineContextSender*)pipeline_context;
mtx_lock(&context->mutex_scanner); ParallelScanner* scanner = parallel_scanner_create(
DirectoryScanner* scanner = directory_scanner_create(
context->config->send_directory, context->config->use_metadata, context->config->chunk_size, context->config->send_directory, context->config->use_metadata, context->config->chunk_size,
context->config->exclude_patterns, context->config->exclude_count, context->config->exclude_patterns, context->config->exclude_count,
context->config->include_patterns, context->config->include_count, context->config->max_size, context->config->include_patterns, context->config->include_count, context->config->max_size,
context->config->min_size, context->config->max_depth); context->config->min_size, 4);
mtx_unlock(&context->mutex_scanner);
Chunk* current_chunk; Chunk* current_chunk;
while ((current_chunk = directory_scanner_next(scanner)) != NULL) { while ((current_chunk = parallel_scanner_next(scanner)) != NULL) {
if (context->config->use_delete) { if (context->config->use_delete) {
mtx_lock(&context->mutex_scanner); mtx_lock(&context->mutex_scanner);
for (int i = 0; i < current_chunk->element_count; i++) { for (int i = 0; i < current_chunk->element_count; i++) {
@@ -390,7 +328,7 @@ static int scan_directory_multithreaded(void* pipeline_context) {
cnd_signal(&context->condition_not_empty_scanner); cnd_signal(&context->condition_not_empty_scanner);
mtx_unlock(&context->mutex_scanner); mtx_unlock(&context->mutex_scanner);
directory_scanner_destroy(scanner); parallel_scanner_destroy(scanner);
return thrd_success; return thrd_success;
} }
@@ -409,9 +347,12 @@ static int load_files_multithreaded(void* pipeline_context) {
} }
if (!context->config->use_sendfile) { if (!context->config->use_sendfile) {
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
if (!file_load_data(chunk->items[i])) { File* f = chunk->items[i];
if (f->data->size > STREAM_THRESHOLD)
continue;
if (!file_load_data(f)) {
log_message(LOG_LEVEL_ERROR, "Failed to load file data, skipping"); log_message(LOG_LEVEL_ERROR, "Failed to load file data, skipping");
file_destroy(chunk->items[i]); file_destroy(f);
chunk->items[i] = NULL; chunk->items[i] = NULL;
} }
} }
@@ -427,15 +368,13 @@ int send_files(Config* config) {
DirectoryScanner* scanner = directory_scanner_create( DirectoryScanner* scanner = directory_scanner_create(
config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns, config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns,
config->exclude_count, config->include_patterns, config->include_count, config->max_size, config->exclude_count, config->include_patterns, config->include_count, config->max_size,
config->min_size, config->max_depth); config->min_size);
Chunk* chunk; Chunk* chunk;
int file_count = 0; int file_count = 0;
unsigned long long total_bytes = 0; unsigned long long total_bytes = 0;
if (!config->quiet)
printf("Dry run: files to be transferred\n"); printf("Dry run: files to be transferred\n");
while ((chunk = directory_scanner_next(scanner)) != NULL) { while ((chunk = directory_scanner_next(scanner)) != NULL) {
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
if (!config->quiet)
printf(" %s (%zu bytes)\n", chunk->items[i]->path, chunk->items[i]->data->size); printf(" %s (%zu bytes)\n", chunk->items[i]->path, chunk->items[i]->data->size);
total_bytes += chunk->items[i]->data->size; total_bytes += chunk->items[i]->data->size;
file_count++; file_count++;
@@ -443,7 +382,6 @@ int send_files(Config* config) {
chunk_destroy(chunk); chunk_destroy(chunk);
} }
directory_scanner_destroy(scanner); directory_scanner_destroy(scanner);
if (!config->quiet)
printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0); printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0);
return 0; return 0;
} }
@@ -480,98 +418,45 @@ int send_files(Config* config) {
client_delete(client); client_delete(client);
return 1; return 1;
} }
g_abort_fd = client->file_descriptor;
struct sigaction sa;
memset(&sa, 0, sizeof(sa));
sa.sa_handler = handle_sigint;
sigaction(SIGINT, &sa, NULL);
sigaction(SIGTERM, &sa, NULL);
DirectoryScanner* scanner = directory_scanner_create( DirectoryScanner* scanner = directory_scanner_create(
config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns, config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns,
config->exclude_count, config->include_patterns, config->include_count, config->max_size, config->exclude_count, config->include_patterns, config->include_count, config->max_size,
config->min_size, config->max_depth); config->min_size);
ArrayList* all_files = array_list_create(NULL);
ArrayList* manifest = config->use_delete ? array_list_create(free) : NULL;
Chunk* current_chunk; Chunk* current_chunk;
unsigned long long total_bytes = 0;
time_t last_progress = 0;
time_t start = time(NULL);
ArrayList* manifest = config->use_delete ? array_list_create(free) : NULL;
while ((current_chunk = directory_scanner_next(scanner)) != NULL) { while ((current_chunk = directory_scanner_next(scanner)) != NULL) {
unsigned long long chunk_bytes = 0;
for (int i = 0; i < current_chunk->element_count; i++) { for (int i = 0; i < current_chunk->element_count; i++) {
File* f = current_chunk->items[i]; chunk_bytes += current_chunk->items[i]->data->size;
array_list_add(all_files, f);
current_chunk->items[i] = NULL;
if (manifest) { if (manifest) {
const char* p = f->path; const char* p = current_chunk->items[i]->path;
if (*p == '/') if (*p == '/')
p++; p++;
array_list_add(manifest, str_dup(p)); array_list_add(manifest, str_dup(p));
} }
} }
chunk_destroy(current_chunk); if (!config->use_sendfile) {
} for (int i = 0; i < current_chunk->element_count; i++) {
directory_scanner_destroy(scanner); File* f = current_chunk->items[i];
scanner = NULL; if (f->data->size > STREAM_THRESHOLD)
bool batch_ok = true;
if (config->use_incremental && all_files->size > 0) {
if (!batch_incremental_check(client, all_files)) {
log_message(LOG_LEVEL_ERROR, "Batch incremental check failed");
batch_ok = false;
}
}
unsigned long long total_bytes = 0;
time_t last_progress = 0;
time_t last_activity = 0;
time_t start = time(NULL);
bool use_sendfile = config->use_sendfile && !config->use_compression;
for (int i = 0; i < all_files->size; i++) {
File* file = (File*)all_files->items[i];
if (file->skip)
continue; continue;
if (g_abort_requested) { if (!file_load_data(f)) {
send_status(client->file_descriptor, STATUS_ABORT);
batch_ok = false;
break;
}
time_t now = time(NULL);
if (now - last_activity >= KEEPALIVE_INTERVAL) {
if (!send_status(client->file_descriptor, STATUS_KEEPALIVE)) {
batch_ok = false;
break;
}
Status s;
if (!receive_status(client->file_descriptor, &s)) {
batch_ok = false;
break;
}
last_activity = now;
}
int compression_level = config->use_compression ? config->compression_level : 0;
if (!send_status(client->file_descriptor, STATUS_NEXT)) {
batch_ok = false;
break;
}
if (use_sendfile) {
if (!file_send_sendfile(file, client->file_descriptor, config->use_metadata, 0, true)) {
log_message(LOG_LEVEL_ERROR, "Failed to send file via sendfile");
batch_ok = false;
break;
}
} else {
if (!file_load_data(file)) {
log_message(LOG_LEVEL_ERROR, "Failed to load file data"); log_message(LOG_LEVEL_ERROR, "Failed to load file data");
continue; continue;
} }
if (!file_send_single_calls(file, client->file_descriptor, config->use_metadata, }
compression_level, true)) { }
log_message(LOG_LEVEL_ERROR, "Failed to send file"); if (send_chunk(client, current_chunk, config) != 0) {
batch_ok = false; log_message(LOG_LEVEL_ERROR, "Failed to send chunk");
chunk_destroy(current_chunk);
break; break;
} }
}
total_bytes += file->data ? file->data->size : 0;
if (config->show_progress) { if (config->show_progress) {
total_bytes += chunk_bytes;
time_t now = time(NULL);
if (now - last_progress >= 1) { if (now - last_progress >= 1) {
last_progress = now; last_progress = now;
double elapsed = difftime(now, start); double elapsed = difftime(now, start);
@@ -580,56 +465,58 @@ int send_files(Config* config) {
fflush(stderr); fflush(stderr);
} }
} }
chunk_destroy(current_chunk);
} }
if (config->use_delete) {
if (batch_ok && config->use_delete && manifest) { if (!send_status(client->file_descriptor, STATUS_MANIFEST)) {
if (!send_status(client->file_descriptor, STATUS_MANIFEST)) array_list_delete(manifest);
batch_ok = false; goto send_fail;
else if (!send_int(client->file_descriptor, manifest->size))
batch_ok = false;
else {
for (int i = 0; i < manifest->size && batch_ok; i++) {
if (!send_str(client->file_descriptor, (char*)manifest->items[i]))
batch_ok = false;
} }
if (!send_int(client->file_descriptor, manifest->size)) {
array_list_delete(manifest);
goto send_fail;
}
for (int i = 0; i < manifest->size; i++) {
if (!send_str(client->file_descriptor, (char*)manifest->items[i])) {
array_list_delete(manifest);
goto send_fail;
} }
} }
array_list_delete(manifest); array_list_delete(manifest);
}
if (batch_ok && !send_status(client->file_descriptor, STATUS_FINISHED)) if (!send_status(client->file_descriptor, STATUS_FINISHED))
batch_ok = false; goto send_fail;
Status s; Status s;
int ok = 0; int ok = receive_status(client->file_descriptor, &s) && s == STATUS_OK;
if (batch_ok)
ok = receive_status(client->file_descriptor, &s) && s == STATUS_OK;
if (config->show_progress) { if (config->show_progress) {
double elapsed = difftime(time(NULL), start); double elapsed = difftime(time(NULL), start);
double rate = elapsed > 0 ? total_bytes / (1048576.0 * elapsed) : 0; double rate = elapsed > 0 ? total_bytes / (1048576.0 * elapsed) : 0;
fprintf(stderr, "\rSent %.1f MB (%.1f MB/s) Done.\n", total_bytes / 1048576.0, rate); fprintf(stderr, "\rSent %.1f MB (%.1f MB/s) Done.\n", total_bytes / 1048576.0, rate);
} }
for (int i = 0; i < all_files->size; i++) directory_scanner_destroy(scanner);
file_destroy(all_files->items[i]);
array_list_delete(all_files);
client_disconnect(client); client_disconnect(client);
client_delete(client); client_delete(client);
return (batch_ok && ok) ? 0 : -1; return ok ? 0 : -1;
send_fail:
directory_scanner_destroy(scanner);
client_disconnect(client);
client_delete(client);
return -1;
} }
int send_files_multithreaded(Config* config) { int send_files_multithreaded(Config* config) {
time_t start_time = time(NULL);
if (config->dry_run) { if (config->dry_run) {
DirectoryScanner* scanner = directory_scanner_create( DirectoryScanner* scanner = directory_scanner_create(
config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns, config->send_directory, config->use_metadata, config->chunk_size, config->exclude_patterns,
config->exclude_count, config->include_patterns, config->include_count, config->max_size, config->exclude_count, config->include_patterns, config->include_count, config->max_size,
config->min_size, config->max_depth); config->min_size);
Chunk* chunk; Chunk* chunk;
int file_count = 0; int file_count = 0;
unsigned long long total_bytes = 0; unsigned long long total_bytes = 0;
if (!config->quiet)
printf("Dry run: files to be transferred\n"); printf("Dry run: files to be transferred\n");
while ((chunk = directory_scanner_next(scanner)) != NULL) { while ((chunk = directory_scanner_next(scanner)) != NULL) {
for (int i = 0; i < chunk->element_count; i++) { for (int i = 0; i < chunk->element_count; i++) {
if (!config->quiet)
printf(" %s (%zu bytes)\n", chunk->items[i]->path, chunk->items[i]->data->size); printf(" %s (%zu bytes)\n", chunk->items[i]->path, chunk->items[i]->data->size);
total_bytes += chunk->items[i]->data->size; total_bytes += chunk->items[i]->data->size;
file_count++; file_count++;
@@ -637,12 +524,22 @@ int send_files_multithreaded(Config* config) {
chunk_destroy(chunk); chunk_destroy(chunk);
} }
directory_scanner_destroy(scanner); directory_scanner_destroy(scanner);
if (!config->quiet)
printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0); printf("Total: %d files, %.1f MB\n", file_count, total_bytes / 1048576.0);
return 0; return 0;
} }
int qsize = config->queue_size > 0 ? config->queue_size : 100; long pages = sysconf(_SC_AVPHYS_PAGES);
long page_size = sysconf(_SC_PAGE_SIZE);
unsigned long long available_memory =
pages > 0 && page_size > 0 ? (unsigned long long)pages * (unsigned long long)page_size
: 512ULL * 1024 * 1024;
unsigned long long avg_file_size = 1024 * 1024;
int qsize = (int)(available_memory / avg_file_size);
if (qsize < 10)
qsize = 10;
if (qsize > 1000)
qsize = 1000;
Queue* q1 = queue_create(qsize, chunk_destroy); Queue* q1 = queue_create(qsize, chunk_destroy);
Queue* q2 = queue_create(qsize, chunk_destroy); Queue* q2 = queue_create(qsize, chunk_destroy);
if (!q1 || !q2) { if (!q1 || !q2) {
@@ -675,12 +572,6 @@ int send_files_multithreaded(Config* config) {
thrd_join(loader, NULL); thrd_join(loader, NULL);
thrd_join(sender, &sender_result); thrd_join(sender, &sender_result);
if (config->stats && !config->quiet) {
double elapsed = difftime(time(NULL), start_time);
printf("\nTransfer statistics:\n");
printf(" Elapsed time: %.1f sec\n", elapsed);
}
pipeline_context_sender_destroy(context); pipeline_context_sender_destroy(context);
return sender_result == thrd_success ? 0 : -1; return sender_result == thrd_success ? 0 : -1;
} }
src/client/scanner.c
+251 -1
View File
@@ -9,6 +9,7 @@
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <sys/stat.h> #include <sys/stat.h>
#include <threads.h>
#include <unistd.h> #include <unistd.h>
typedef struct { typedef struct {
@@ -79,7 +80,6 @@ static Chunk* chunk_data_to_chunk(ArrayList* chunk_data) {
return chunk; return chunk;
} }
// Returns: 1 on success, 0 if no more directories in queue, -1 on opendir failure
static int open_next_directory(DirectoryScanner* scanner) { static int open_next_directory(DirectoryScanner* scanner) {
if (scanner->current_dir) { if (scanner->current_dir) {
closedir(scanner->current_dir); closedir(scanner->current_dir);
@@ -202,3 +202,253 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) {
array_list_delete(chunk_data); array_list_delete(chunk_data);
return NULL; return NULL;
} }
typedef struct {
ParallelScanner* ps;
char** dirs;
int dir_count;
bool use_metadata;
unsigned long long chunk_size;
char** exclude_patterns;
int exclude_count;
char** include_patterns;
int include_count;
unsigned long long max_size;
unsigned long long min_size;
} ParallelWorkerArg;
static int parallel_worker_thread(void* arg) {
ParallelWorkerArg* wa = (ParallelWorkerArg*)arg;
for (int i = 0; i < wa->dir_count; i++) {
DirectoryScanner* ds = directory_scanner_create(
wa->dirs[i], wa->use_metadata, wa->chunk_size, wa->exclude_patterns, wa->exclude_count,
wa->include_patterns, wa->include_count, wa->max_size, wa->min_size);
Chunk* chunk;
while ((chunk = directory_scanner_next(ds)) != NULL) {
queue_enqueue_multithreaded(wa->ps->result_queue, chunk, &wa->ps->result_mutex,
&wa->ps->result_not_empty, &wa->ps->result_not_full);
}
directory_scanner_destroy(ds);
free(wa->dirs[i]);
}
ParallelScanner* ps = wa->ps;
free(wa->dirs);
free(wa);
mtx_lock(&ps->result_mutex);
ps->completed++;
if (ps->completed >= ps->num_threads) {
ps->done = true;
cnd_signal(&ps->result_not_empty);
}
mtx_unlock(&ps->result_mutex);
return thrd_success;
}
ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata,
unsigned long long chunk_size, char** exclude_patterns,
int exclude_count, char** include_patterns,
int include_count, unsigned long long max_size,
unsigned long long min_size, int num_threads) {
ParallelScanner* ps = calloc(1, sizeof(ParallelScanner));
if (!ps)
return NULL;
ps->result_queue = queue_create(100, chunk_destroy);
if (!ps->result_queue) {
free(ps);
return NULL;
}
if (mtx_init(&ps->result_mutex, mtx_plain) != thrd_success ||
cnd_init(&ps->result_not_empty) != thrd_success ||
cnd_init(&ps->result_not_full) != thrd_success) {
queue_destroy(ps->result_queue);
free(ps);
return NULL;
}
DIR* dir = opendir(root_directory);
if (!dir) {
perror("Could not open root directory for parallel scan");
parallel_scanner_destroy(ps);
return NULL;
}
ArrayList* root_files = array_list_create(file_destroy);
ArrayList* subdirs = array_list_create(free);
struct dirent* entry;
while ((entry = readdir(dir)) != NULL) {
if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
continue;
char* cur_path = path_cat(root_directory, entry->d_name);
if (!cur_path)
continue;
struct stat st;
if (stat(cur_path, &st) != 0) {
free(cur_path);
continue;
}
if (S_ISDIR(st.st_mode)) {
array_list_add(subdirs, cur_path);
} else {
bool excluded = false;
for (int i = 0; i < exclude_count; i++) {
if (glob_match(exclude_patterns[i], entry->d_name)) {
excluded = true;
break;
}
}
if (excluded) {
free(cur_path);
continue;
}
if (include_count > 0) {
bool included = false;
for (int i = 0; i < include_count; i++) {
if (glob_match(include_patterns[i], entry->d_name)) {
included = true;
break;
}
}
if (!included) {
free(cur_path);
continue;
}
}
if ((max_size > 0 && (unsigned long long)st.st_size > max_size) ||
(min_size > 0 && (unsigned long long)st.st_size < min_size)) {
free(cur_path);
continue;
}
File* file = file_create(cur_path);
free(cur_path);
if (!file)
continue;
file->data->size = st.st_size;
if (use_metadata)
file->metadata = file_metadata_create(&st);
array_list_add(root_files, file);
}
}
closedir(dir);
unsigned long long cs = chunk_size > 0 ? chunk_size : DESIRED_CHUNK_SIZE;
if (root_files->size > 0) {
ArrayList* batch = array_list_create(NULL);
unsigned long long batch_size = 0;
Chunk* first = NULL;
for (int i = 0; i < root_files->size; i++) {
File* f = (File*)root_files->items[i];
array_list_add(batch, f);
batch_size += f->data->size;
if (batch_size >= cs || i == root_files->size - 1) {
void** items = array_list_to_array(batch);
Chunk* c = chunk_create((File**)items, batch->size);
free(items);
batch->item_destroyer = NULL;
array_list_delete(batch);
batch = NULL;
if (!first) {
first = c;
} else {
queue_enqueue_multithreaded(ps->result_queue, c, &ps->result_mutex, &ps->result_not_empty,
&ps->result_not_full);
}
if (i < root_files->size - 1) {
batch = array_list_create(NULL);
batch_size = 0;
}
}
}
if (batch) {
batch->item_destroyer = NULL;
array_list_delete(batch);
}
ps->initial_chunk = first;
root_files->item_destroyer = NULL;
}
array_list_delete(root_files);
int n = num_threads > 0 ? num_threads : 4;
if (n > subdirs->size)
n = subdirs->size > 0 ? subdirs->size : 1;
if (subdirs->size > 0) {
ps->num_threads = n;
ps->threads = calloc(n, sizeof(thrd_t));
if (!ps->threads) {
array_list_delete(subdirs);
parallel_scanner_destroy(ps);
return NULL;
}
int dirs_per_thread = subdirs->size / n;
int remainder = subdirs->size % n;
int start = 0;
for (int t = 0; t < n; t++) {
int count = dirs_per_thread + (t < remainder ? 1 : 0);
if (count == 0)
break;
ParallelWorkerArg* wa = calloc(1, sizeof(ParallelWorkerArg));
if (!wa)
break;
wa->ps = ps;
wa->dirs = calloc(count, sizeof(char*));
if (!wa->dirs) {
free(wa);
break;
}
for (int j = 0; j < count; j++)
wa->dirs[j] = str_dup((char*)subdirs->items[start + j]);
wa->dir_count = count;
wa->use_metadata = use_metadata;
wa->chunk_size = cs;
wa->exclude_patterns = exclude_patterns;
wa->exclude_count = exclude_count;
wa->include_patterns = include_patterns;
wa->include_count = include_count;
wa->max_size = max_size;
wa->min_size = min_size;
start += count;
if (thrd_create(&ps->threads[t], parallel_worker_thread, wa) != thrd_success) {
for (int j = 0; j < count; j++)
free(wa->dirs[j]);
free(wa->dirs);
free(wa);
ps->num_threads = t;
break;
}
}
}
array_list_delete(subdirs);
return ps;
}
Chunk* parallel_scanner_next(ParallelScanner* ps) {
if (ps->initial_chunk) {
Chunk* c = ps->initial_chunk;
ps->initial_chunk = NULL;
return c;
}
if (ps->num_threads == 0) {
ps->done = true;
return NULL;
}
Chunk* chunk = queue_dequeue_multithreaded(
ps->result_queue, &ps->result_mutex, &ps->result_not_empty, &ps->result_not_full, &ps->done);
return chunk;
}
void parallel_scanner_destroy(ParallelScanner* ps) {
if (!ps)
return;
ps->done = true;
cnd_signal(&ps->result_not_empty);
for (int i = 0; i < ps->num_threads; i++)
thrd_join(ps->threads[i], NULL);
free(ps->threads);
if (ps->initial_chunk)
chunk_destroy(ps->initial_chunk);
queue_destroy(ps->result_queue);
mtx_destroy(&ps->result_mutex);
cnd_destroy(&ps->result_not_empty);
cnd_destroy(&ps->result_not_full);
free(ps);
}
src/client/scanner.h
+23 -4
View File
@@ -5,6 +5,7 @@
#include "queue.h" #include "queue.h"
#include <dirent.h> #include <dirent.h>
#include <stdbool.h> #include <stdbool.h>
#include <threads.h>
typedef struct { typedef struct {
Queue* directories; Queue* directories;
@@ -18,16 +19,34 @@ typedef struct {
int include_count; int include_count;
unsigned long long max_size; unsigned long long max_size;
unsigned long long min_size; unsigned long long min_size;
int max_depth;
int current_depth;
} DirectoryScanner; } DirectoryScanner;
DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_metadata, typedef struct {
Queue* result_queue;
mtx_t result_mutex;
cnd_t result_not_empty;
cnd_t result_not_full;
int num_threads;
thrd_t* threads;
bool done;
int completed;
Chunk* initial_chunk;
} ParallelScanner;
DirectoryScanner* directory_scanner_create(char* root_directory, bool use_metadata,
unsigned long long chunk_size, char** exclude_patterns, unsigned long long chunk_size, char** exclude_patterns,
int exclude_count, char** include_patterns, int exclude_count, char** include_patterns,
int include_count, unsigned long long max_size, int include_count, unsigned long long max_size,
unsigned long long min_size, int max_depth); unsigned long long min_size);
Chunk* directory_scanner_next(DirectoryScanner* scanner); Chunk* directory_scanner_next(DirectoryScanner* scanner);
void directory_scanner_destroy(DirectoryScanner* scanner); void directory_scanner_destroy(DirectoryScanner* scanner);
ParallelScanner* parallel_scanner_create(char* root_directory, bool use_metadata,
unsigned long long chunk_size, char** exclude_patterns,
int exclude_count, char** include_patterns,
int include_count, unsigned long long max_size,
unsigned long long min_size, int num_threads);
Chunk* parallel_scanner_next(ParallelScanner* scanner);
void parallel_scanner_destroy(ParallelScanner* scanner);
#endif #endif
src/shared/compression.c
+21 -24
View File
@@ -1,26 +1,31 @@
#include "compression.h" #include "compression.h"
#include "data.h" #include "data.h"
#include "log.h" #include "log.h"
#include <stdint.h> #include "stdlib.h"
#include <stdlib.h> #include "string.h"
#include <strings.h>
#include "zstd.h" #include "zstd.h"
#define INITIAL_DECOMPRESS_BUF_SIZE (1024 * 1024) #define INITIAL_DECOMPRESS_BUF_SIZE (1024 * 1024)
static const char* SKIP_COMPRESSION_EXTENSIONS[] = {".jpg", ".jpeg", ".png", ".gif", ".mp4", ".mkv",
".zip", ".gz", ".xz", ".zst", NULL};
bool compression_should_skip(const char* path) {
if (!path)
return false;
const char* dot = strrchr(path, '.');
if (!dot)
return false;
for (int i = 0; SKIP_COMPRESSION_EXTENSIONS[i]; i++) {
if (strcasecmp(dot, SKIP_COMPRESSION_EXTENSIONS[i]) == 0)
return true;
}
return false;
}
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");
/* 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); size_t dst_size = ZSTD_compressBound(data_to_compress->size);
Data* compressed_data = data_create_empty(dst_size); Data* compressed_data = data_create_empty(dst_size);
if (compressed_data == NULL) if (compressed_data == NULL)
@@ -79,16 +84,8 @@ Data* data_decompress(Data* compressed_data) {
return NULL; return NULL;
} }
size_t buf_size = INITIAL_DECOMPRESS_BUF_SIZE; size_t buf_size =
if (!ZSTD_isError(dst_size) && dst_size > 0) { (!ZSTD_isError(dst_size) && dst_size > 0) ? (size_t)dst_size : INITIAL_DECOMPRESS_BUF_SIZE;
if (dst_size > SIZE_MAX) {
log_message(LOG_LEVEL_ERROR,
"Decompressed size %llu exceeds addressable memory, using fallback buffer",
dst_size);
} else {
buf_size = (size_t)dst_size;
}
}
Data* uncompressed_data = data_create_empty(buf_size); Data* uncompressed_data = data_create_empty(buf_size);
if (!uncompressed_data) { if (!uncompressed_data) {
log_message(LOG_LEVEL_ERROR, "Failed to allocate decompression buffer"); log_message(LOG_LEVEL_ERROR, "Failed to allocate decompression buffer");
src/shared/compression.h
+2
View File
@@ -2,8 +2,10 @@
#define COMPRESSION_H #define COMPRESSION_H
#include "data.h" #include "data.h"
#include <stdbool.h>
Data* data_compress(Data* data_to_compress, int compression_level); Data* data_compress(Data* data_to_compress, int compression_level);
Data* data_decompress(Data* compressed_data); Data* data_decompress(Data* compressed_data);
bool compression_should_skip(const char* path);
#endif #endif
src/shared/file.c
+1 -1
View File
@@ -104,7 +104,7 @@ bool file_send_single_calls(File* file, int file_descriptor, bool use_metadata,
int compression_level, bool send_path) { int compression_level, bool send_path) {
const Data* data_to_send = file->data; const Data* data_to_send = file->data;
Data* compressed_data = NULL; Data* compressed_data = NULL;
if (compression_level > 0) { if (compression_level > 0 && !compression_should_skip(file->path)) {
compressed_data = data_compress(file->data, compression_level); compressed_data = data_compress(file->data, compression_level);
if (compressed_data == NULL) { if (compressed_data == NULL) {
log_message(LOG_LEVEL_ERROR, "Failed to compress file data"); log_message(LOG_LEVEL_ERROR, "Failed to compress file data");
src/shared/protocol.c
+7 -7
View File
@@ -2,6 +2,7 @@
#include "log.h" #include "log.h"
#include <errno.h> #include <errno.h>
#include <openssl/ssl.h> #include <openssl/ssl.h>
#include <poll.h>
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
@@ -14,7 +15,7 @@
static __thread int io_read_fd = -1; static __thread int io_read_fd = -1;
static __thread int io_write_fd = -1; static __thread int io_write_fd = -1;
static SSL* io_ssl = NULL; static SSL* io_ssl;
static unsigned long long io_bwlimit = 0; static unsigned long long io_bwlimit = 0;
static long long bw_tokens = 0; static long long bw_tokens = 0;
@@ -52,12 +53,11 @@ static void bw_throttle(size_t bytes_written) {
bw_tokens -= (long long)bytes_written; bw_tokens -= (long long)bytes_written;
if (bw_tokens < 0) { if (bw_tokens < 0) {
long long deficit_ns = (long long)((double)(-bw_tokens) / io_bwlimit * 1000000000.0); long long deficit_us = (long long)((double)(-bw_tokens) / io_bwlimit * 1000000.0);
struct timespec sleep_time, remaining; if (deficit_us >= 1000)
sleep_time.tv_sec = deficit_ns / 1000000000LL; poll(NULL, 0, (int)(deficit_us / 1000));
sleep_time.tv_nsec = deficit_ns % 1000000000LL; else
while (nanosleep(&sleep_time, &remaining) < 0 && errno == EINTR) usleep((useconds_t)deficit_us);
sleep_time = remaining;
bw_tokens = 0; bw_tokens = 0;
clock_gettime(CLOCK_MONOTONIC, &bw_last_refill); clock_gettime(CLOCK_MONOTONIC, &bw_last_refill);
} }