diff --git a/src/client/client_send.c b/src/client/client_send.c index d01e2d3..76872bd 100644 --- a/src/client/client_send.c +++ b/src/client/client_send.c @@ -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; diff --git a/src/client/scanner.c b/src/client/scanner.c index de4f647..24413c8 100644 --- a/src/client/scanner.c +++ b/src/client/scanner.c @@ -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++; diff --git a/src/shared/transport_tcp.c b/src/shared/transport_tcp.c index 099294d..062d60c 100644 --- a/src/shared/transport_tcp.c +++ b/src/shared/transport_tcp.c @@ -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; diff --git a/src/shared/transport_tcp.h b/src/shared/transport_tcp.h index 8024c92..156531a 100644 --- a/src/shared/transport_tcp.h +++ b/src/shared/transport_tcp.h @@ -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 diff --git a/src/shared/transport_tls.c b/src/shared/transport_tls.c index b8c53b3..2c5a4ff 100644 --- a/src/shared/transport_tls.c +++ b/src/shared/transport_tls.c @@ -9,6 +9,7 @@ #include #include #include +#include #include 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();