refactor: split transfer and protocol responsibilities #212
@@ -658,6 +658,7 @@ int send_files(Config* config) {
|
|||||||
if (config->use_delete) {
|
if (config->use_delete) {
|
||||||
if (send_delete_manifest(client->file_descriptor, manifest) != 0) {
|
if (send_delete_manifest(client->file_descriptor, manifest) != 0) {
|
||||||
array_list_delete(manifest);
|
array_list_delete(manifest);
|
||||||
|
manifest = NULL;
|
||||||
goto send_fail;
|
goto send_fail;
|
||||||
}
|
}
|
||||||
array_list_delete(manifest);
|
array_list_delete(manifest);
|
||||||
|
|||||||
@@ -667,6 +667,11 @@ Chunk* parallel_scanner_next(ParallelScanner* ps) {
|
|||||||
}
|
}
|
||||||
if (ps->num_threads == 0) {
|
if (ps->num_threads == 0) {
|
||||||
mtx_lock(&ps->result_mutex);
|
mtx_lock(&ps->result_mutex);
|
||||||
|
if (!queue_is_empty(ps->result_queue)) {
|
||||||
|
Chunk* chunk = queue_dequeue(ps->result_queue);
|
||||||
|
mtx_unlock(&ps->result_mutex);
|
||||||
|
return chunk;
|
||||||
|
}
|
||||||
ps->done = true;
|
ps->done = true;
|
||||||
mtx_unlock(&ps->result_mutex);
|
mtx_unlock(&ps->result_mutex);
|
||||||
return NULL;
|
return NULL;
|
||||||
|
|||||||
@@ -106,6 +106,7 @@ void handler(int file_descriptor) {
|
|||||||
protocol_session_unbind();
|
protocol_session_unbind();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
context->session.total_allocated_bytes = session.total_allocated_bytes;
|
||||||
thrd_t receiver, writer;
|
thrd_t receiver, writer;
|
||||||
bool receiver_created = thrd_create(&receiver, receive_thread, context) == thrd_success;
|
bool receiver_created = thrd_create(&receiver, receive_thread, context) == thrd_success;
|
||||||
bool writer_created = false;
|
bool writer_created = false;
|
||||||
|
|||||||
@@ -385,6 +385,34 @@ static void test_scanner_no_patterns() {
|
|||||||
rmdir(dir);
|
rmdir(dir);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static void test_parallel_scanner_root_chunks_without_workers() {
|
||||||
|
const char* dir = "test_parallel_scan_root";
|
||||||
|
const char* file1 = "test_parallel_scan_root/a.txt";
|
||||||
|
const char* file2 = "test_parallel_scan_root/b.txt";
|
||||||
|
|
||||||
|
EXPECT_EQ_INT(mkdir(dir, 0755), 0);
|
||||||
|
create_test_file(file1, "a");
|
||||||
|
create_test_file(file2, "b");
|
||||||
|
|
||||||
|
ParallelScanner* scanner = parallel_scanner_create(dir, false, 1, NULL, 0, NULL, 0, 0, 0, 0, 0,
|
||||||
|
false, false, false, false, false);
|
||||||
|
EXPECT_NOT_NULL(scanner);
|
||||||
|
|
||||||
|
int total_files = 0;
|
||||||
|
Chunk* chunk;
|
||||||
|
while ((chunk = parallel_scanner_next(scanner)) != NULL) {
|
||||||
|
total_files += chunk->element_count;
|
||||||
|
chunk_destroy(chunk);
|
||||||
|
}
|
||||||
|
EXPECT_EQ_INT(total_files, 2);
|
||||||
|
EXPECT_FALSE(parallel_scanner_failed(scanner));
|
||||||
|
|
||||||
|
parallel_scanner_destroy(scanner);
|
||||||
|
unlink(file1);
|
||||||
|
unlink(file2);
|
||||||
|
rmdir(dir);
|
||||||
|
}
|
||||||
|
|
||||||
void test_scanner() {
|
void test_scanner() {
|
||||||
test_scanner_single_file();
|
test_scanner_single_file();
|
||||||
test_scanner_multiple_files();
|
test_scanner_multiple_files();
|
||||||
@@ -399,4 +427,5 @@ void test_scanner() {
|
|||||||
test_scanner_size_range();
|
test_scanner_size_range();
|
||||||
test_scanner_mixed_patterns();
|
test_scanner_mixed_patterns();
|
||||||
test_scanner_no_patterns();
|
test_scanner_no_patterns();
|
||||||
|
test_parallel_scanner_root_chunks_without_workers();
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user