refactor: split transfer and protocol responsibilities #212

Open
TapTap wants to merge 11 commits from refactor/codebase-structure into dev
5 changed files with 31 additions and 14 deletions
Showing only changes of commit 604a14f0be - Show all commits
+6
View File
@@ -378,6 +378,7 @@ static int send_chunks_multithreaded(void* pipeline_context) {
if (context->config->transport == TRANSPORT_TCP)
fprintf(stderr, "Error: could not connect to server%s\n",
context->config->use_tls ? " via TLS" : "");
pipeline_cancel(context);
mark_sender_done(context);
return thrd_error;
}
@@ -386,6 +387,7 @@ static int send_chunks_multithreaded(void* pipeline_context) {
protocol_session_set_ssl(&session, (SSL*)client->ssl);
protocol_session_bind(&session);
if (!config_send(client->file_descriptor, context->config)) {
pipeline_cancel(context);
disconnect_transfer_client(client);
mark_sender_done(context);
protocol_session_unbind();
@@ -713,6 +715,10 @@ int send_files_multithreaded(Config* config) {
}
if (config->use_delete)
context->manifest = create_transfer_manifest(config);
if (config->use_delete && !context->manifest) {
pipeline_context_sender_destroy(context);
return 1;
}
thrd_t scanner, loader, sender;
bool scanner_created = false;
+16 -13
View File
@@ -400,6 +400,18 @@ static int parallel_worker_thread(void* arg) {
return thrd_success;
}
static void parallel_scanner_creation_failed(ParallelScanner* ps) {
mtx_lock(&ps->result_mutex);
ps->failed = true;
atomic_store(&ps->cancelled, true);
ps->expected_threads = ps->created_threads;
if (ps->completed >= ps->expected_threads)
ps->done = true;
cnd_broadcast(&ps->result_not_empty);
cnd_broadcast(&ps->result_not_full);
mtx_unlock(&ps->result_mutex);
}
ParallelScanner* parallel_scanner_create_with_options(const char* root_directory,
const ScannerOptions* options) {
if (!root_directory || !options)
@@ -589,16 +601,14 @@ ParallelScanner* parallel_scanner_create_with_options(const char* root_directory
break;
ParallelWorkerArg* wa = calloc(1, sizeof(ParallelWorkerArg));
if (!wa) {
ps->failed = true;
ps->expected_threads = ps->created_threads;
parallel_scanner_creation_failed(ps);
break;
}
wa->ps = ps;
wa->dirs = calloc(count, sizeof(char*));
if (!wa->dirs) {
free(wa);
ps->failed = true;
ps->expected_threads = ps->created_threads;
parallel_scanner_creation_failed(ps);
break;
}
bool dup_ok = true;
@@ -612,8 +622,7 @@ ParallelScanner* parallel_scanner_create_with_options(const char* root_directory
free(wa->dirs[j]);
free(wa->dirs);
free(wa);
ps->failed = true;
ps->expected_threads = ps->created_threads;
parallel_scanner_creation_failed(ps);
break;
}
wa->dir_count = count;
@@ -625,13 +634,7 @@ ParallelScanner* parallel_scanner_create_with_options(const char* root_directory
free(wa->dirs[j]);
free(wa->dirs);
free(wa);
ps->failed = true;
atomic_store(&ps->cancelled, true);
ps->expected_threads = ps->created_threads;
mtx_lock(&ps->result_mutex);
cnd_broadcast(&ps->result_not_empty);
cnd_broadcast(&ps->result_not_full);
mtx_unlock(&ps->result_mutex);
parallel_scanner_creation_failed(ps);
break;
}
ps->num_threads++;
+4
View File
@@ -151,6 +151,10 @@ int tcp_get_contimeout_sec(void) {
return g_contimeout_sec;
}
int tcp_get_timeout_sec(void) {
return g_timeout_sec;
}
static void tcp_apply_socket_timeout(int fd) {
struct timeval tv;
tv.tv_sec = g_timeout_sec;
+1
View File
@@ -35,5 +35,6 @@ void client_disconnect(Client* client);
void client_delete(Client* client);
void tcp_set_timeouts(int timeout_sec, int contimeout_sec);
int tcp_get_contimeout_sec(void);
int tcp_get_timeout_sec(void);
#endif
+4 -1
View File
@@ -9,6 +9,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
bool tls_global_init(void) {
@@ -92,6 +93,7 @@ static SSL* wrap_fd_with_ssl(int fd, SSL_CTX* ctx, bool is_server, const char* h
}
// Retry SSL_accept/SSL_connect on WANT_READ/WANT_WRITE (non-blocking handshake)
time_t deadline = time(NULL) + (is_server ? tcp_get_timeout_sec() : tcp_get_contimeout_sec());
int ret;
do {
if (is_server)
@@ -101,7 +103,8 @@ static SSL* wrap_fd_with_ssl(int fd, SSL_CTX* ctx, bool is_server, const char* h
if (ret <= 0) {
int ssl_err = SSL_get_error(ssl, ret);
if (ssl_err == SSL_ERROR_WANT_READ || ssl_err == SSL_ERROR_WANT_WRITE)
if ((ssl_err == SSL_ERROR_WANT_READ || ssl_err == SSL_ERROR_WANT_WRITE) &&
time(NULL) < deadline)
continue;
log_message(LOG_LEVEL_ERROR, "SSL %s failed", is_server ? "accept" : "connect");
log_ssl_errors();