refactor: split parallel_scanner_create_with_options into focused helpers

- parallel_scanner_init(): result queue + sync primitive setup with unwinding
- batch_files(): root-file chunk batching, reusable by other scan paths
- scan_root_directory()/scan_root_entry(): root-dir scanning
- spawn_parallel_workers(): worker thread creation with per-thread arg setup
Main function reduced from ~230 to ~40 lines
This commit is contained in:
2026-08-30 13:56:27 +02:00
parent e3e766ba3d
commit 3fa3e150ce
+134 -96
View File
@@ -412,18 +412,11 @@ static void parallel_scanner_creation_failed(ParallelScanner* ps) {
mtx_unlock(&ps->result_mutex); mtx_unlock(&ps->result_mutex);
} }
ParallelScanner* parallel_scanner_create_with_options(const char* root_directory, /* Initialize result queue and synchronization primitives. Returns true on success. */
const ScannerOptions* options) { static bool parallel_scanner_init(ParallelScanner* ps) {
if (!root_directory || !options)
return NULL;
ParallelScanner* ps = calloc(1, sizeof(ParallelScanner));
if (!ps)
return NULL;
ps->result_queue = queue_create(100, chunk_destroy); ps->result_queue = queue_create(100, chunk_destroy);
if (!ps->result_queue) { if (!ps->result_queue)
free(ps); return false;
return NULL;
}
atomic_init(&ps->cancelled, false); atomic_init(&ps->cancelled, false);
int init = 0; int init = 0;
bool ok = true; bool ok = true;
@@ -448,92 +441,37 @@ ParallelScanner* parallel_scanner_create_with_options(const char* root_directory
if (init >= 1) if (init >= 1)
mtx_destroy(&ps->result_mutex); mtx_destroy(&ps->result_mutex);
queue_destroy(ps->result_queue); queue_destroy(ps->result_queue);
free(ps); ps->result_queue = NULL;
return NULL; return false;
}
return true;
} }
DIR* dir = opendir(root_directory); /* Split files into chunks of roughly chunk_size bytes. Returns the first chunk (also stored
if (!dir) { * chunks beyond the first are enqueued on `queue`). Nulls out consumed entries in `files`.
perror("Could not open root directory for parallel scan"); * Sets *failed on allocation/enqueue errors. */
parallel_scanner_destroy(ps); static Chunk* batch_files(ArrayList* files, unsigned long long chunk_size, Queue* queue,
bool* failed) {
Chunk* first = NULL;
if (files->size <= 0)
return NULL; return NULL;
}
ArrayList* root_files = array_list_create(file_destroy);
ArrayList* subdirs = array_list_create(free);
if (!root_files || !subdirs) {
array_list_delete(root_files);
array_list_delete(subdirs);
closedir(dir);
parallel_scanner_destroy(ps);
return NULL;
}
const struct dirent* entry;
while ((entry = readdir(dir)) != NULL) {
if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
continue;
ScannerEntry inspected;
int inspection =
scanner_inspect_entry(options, root_directory, root_directory, entry->d_name, &inspected);
if (inspection < 0) {
ps->failed = true;
continue;
}
if (inspection == 0)
continue;
char* cur_path = inspected.path;
struct stat st = inspected.stats;
if (inspected.is_directory) {
if (!array_list_add(subdirs, cur_path)) {
free(cur_path);
ps->failed = true;
}
} else {
File* file = file_create(cur_path);
free(cur_path);
if (!file) {
ps->failed = true;
continue;
}
file->data->size = st.st_size;
if (options->use_metadata)
file->metadata = file_metadata_create(&st);
if (options->use_metadata && !file->metadata) {
file_destroy(file);
ps->failed = true;
continue;
}
if (!array_list_add(root_files, file)) {
file_destroy(file);
ps->failed = true;
}
}
}
closedir(dir);
unsigned long long cs = options->chunk_size > 0 ? options->chunk_size : DESIRED_CHUNK_SIZE;
if (root_files->size > 0) {
ArrayList* batch = array_list_create(NULL); ArrayList* batch = array_list_create(NULL);
if (!batch) { if (!batch) {
ps->failed = true; *failed = true;
array_list_delete(root_files);
array_list_delete(subdirs);
parallel_scanner_destroy(ps);
return NULL; return NULL;
} }
unsigned long long batch_size = 0; unsigned long long batch_size = 0;
Chunk* first = NULL; for (int i = 0; i < files->size; i++) {
for (int i = 0; i < root_files->size; i++) { File* f = (File*)files->items[i];
File* f = (File*)root_files->items[i];
if (!array_list_add(batch, f)) { if (!array_list_add(batch, f)) {
ps->failed = true; *failed = true;
break; break;
} }
batch_size += f->data->size; batch_size += f->data->size;
if (batch_size >= cs || i == root_files->size - 1) { if (batch_size >= chunk_size || i == files->size - 1) {
void** items = array_list_to_array(batch); void** items = array_list_to_array(batch);
if (!items) { if (!items) {
ps->failed = true; *failed = true;
array_list_delete(batch); array_list_delete(batch);
batch = NULL; batch = NULL;
break; break;
@@ -541,29 +479,29 @@ ParallelScanner* parallel_scanner_create_with_options(const char* root_directory
Chunk* c = chunk_create((File**)items, batch->size); Chunk* c = chunk_create((File**)items, batch->size);
free(items); free(items);
if (!c) { if (!c) {
ps->failed = true; *failed = true;
array_list_delete(batch); array_list_delete(batch);
batch = NULL; batch = NULL;
break; break;
} }
int batch_start = i - batch->size + 1; int batch_start = i - batch->size + 1;
for (int j = batch_start; j <= i; j++) for (int j = batch_start; j <= i; j++)
root_files->items[j] = NULL; files->items[j] = NULL;
batch->item_destroyer = NULL; batch->item_destroyer = NULL;
array_list_delete(batch); array_list_delete(batch);
batch = NULL; batch = NULL;
if (!first) { if (!first) {
first = c; first = c;
} else { } else {
if (!queue_enqueue(ps->result_queue, c)) { if (!queue_enqueue(queue, c)) {
chunk_destroy(c); chunk_destroy(c);
ps->failed = true; *failed = true;
} }
} }
if (i < root_files->size - 1) { if (i < files->size - 1) {
batch = array_list_create(NULL); batch = array_list_create(NULL);
if (!batch) { if (!batch) {
ps->failed = true; *failed = true;
break; break;
} }
batch_size = 0; batch_size = 0;
@@ -574,22 +512,88 @@ ParallelScanner* parallel_scanner_create_with_options(const char* root_directory
batch->item_destroyer = NULL; batch->item_destroyer = NULL;
array_list_delete(batch); array_list_delete(batch);
} }
ps->initial_chunk = first; return first;
} }
array_list_delete(root_files);
/* Scan one root-directory entry into either the subdirs or files list. */
static void scan_root_entry(const ScannerOptions* options, const char* root_directory,
const struct dirent* entry, ArrayList* root_files,
ArrayList* subdirs, ParallelScanner* ps) {
ScannerEntry inspected;
int inspection =
scanner_inspect_entry(options, root_directory, root_directory, entry->d_name, &inspected);
if (inspection < 0) {
ps->failed = true;
return;
}
if (inspection == 0)
return;
char* cur_path = inspected.path;
struct stat st = inspected.stats;
if (inspected.is_directory) {
if (!array_list_add(subdirs, cur_path)) {
free(cur_path);
ps->failed = true;
}
return;
}
File* file = file_create(cur_path);
free(cur_path);
if (!file) {
ps->failed = true;
return;
}
file->data->size = st.st_size;
if (options->use_metadata)
file->metadata = file_metadata_create(&st);
if (options->use_metadata && !file->metadata) {
file_destroy(file);
ps->failed = true;
return;
}
if (!array_list_add(root_files, file)) {
file_destroy(file);
ps->failed = true;
}
}
/* Scan the root directory itself, collecting root files and subdirectories.
* Returns false if the root directory could not be opened. */
static bool scan_root_directory(ParallelScanner* ps, const char* root_directory,
const ScannerOptions* options, ArrayList* root_files,
ArrayList* subdirs) {
DIR* dir = opendir(root_directory);
if (!dir) {
perror("Could not open root directory for parallel scan");
return false;
}
const struct dirent* entry;
while ((entry = readdir(dir)) != NULL) {
if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
continue;
scan_root_entry(options, root_directory, entry, root_files, subdirs, ps);
}
closedir(dir);
return true;
}
/* Spawn worker threads, one per group of subdirectories. */
static void spawn_parallel_workers(ParallelScanner* ps, ArrayList* subdirs,
const ScannerOptions* options, unsigned long long cs) {
if (subdirs->size <= 0)
return;
int n = options->num_threads > 0 ? options->num_threads : 4; int n = options->num_threads > 0 ? options->num_threads : 4;
if (n > subdirs->size) if (n > subdirs->size)
n = subdirs->size > 0 ? subdirs->size : 1; n = subdirs->size;
if (subdirs->size > 0) {
ps->num_threads = n; ps->num_threads = n;
ps->expected_threads = n; ps->expected_threads = n;
ps->threads = calloc(n, sizeof(thrd_t)); ps->threads = calloc(n, sizeof(thrd_t));
if (!ps->threads) { if (!ps->threads) {
array_list_delete(subdirs); ps->num_threads = 0;
parallel_scanner_destroy(ps); ps->expected_threads = 0;
return NULL; ps->failed = true;
return;
} }
int dirs_per_thread = subdirs->size / n; int dirs_per_thread = subdirs->size / n;
int remainder = subdirs->size % n; int remainder = subdirs->size % n;
@@ -641,6 +645,40 @@ ParallelScanner* parallel_scanner_create_with_options(const char* root_directory
ps->created_threads++; ps->created_threads++;
} }
} }
ParallelScanner* parallel_scanner_create_with_options(const char* root_directory,
const ScannerOptions* options) {
if (!root_directory || !options)
return NULL;
ParallelScanner* ps = calloc(1, sizeof(ParallelScanner));
if (!ps)
return NULL;
if (!parallel_scanner_init(ps)) {
free(ps);
return NULL;
}
ArrayList* root_files = array_list_create(file_destroy);
ArrayList* subdirs = array_list_create(free);
if (!root_files || !subdirs) {
array_list_delete(root_files);
array_list_delete(subdirs);
parallel_scanner_destroy(ps);
return NULL;
}
if (!scan_root_directory(ps, root_directory, options, root_files, subdirs)) {
array_list_delete(root_files);
array_list_delete(subdirs);
parallel_scanner_destroy(ps);
return NULL;
}
unsigned long long cs = options->chunk_size > 0 ? options->chunk_size : DESIRED_CHUNK_SIZE;
ps->initial_chunk = batch_files(root_files, cs, ps->result_queue, &ps->failed);
array_list_delete(root_files);
spawn_parallel_workers(ps, subdirs, options, cs);
array_list_delete(subdirs); array_list_delete(subdirs);
return ps; return ps;
} }