From 320bb44b10495eb50efcb77a9799ae5f5f1ada51 Mon Sep 17 00:00:00 2001 From: Andrew D Smith Date: Tue, 6 Oct 2026 17:07:44 -0700 Subject: [PATCH] src/falco_analyzer.cpp: catching exceptions within threads for easier debugging --- src/falco_analyzer.cpp | 94 ++++++++++++++++++++++++------------------ 1 file changed, 54 insertions(+), 40 deletions(-) diff --git a/src/falco_analyzer.cpp b/src/falco_analyzer.cpp index e17a965..1b9bd88 100644 --- a/src/falco_analyzer.cpp +++ b/src/falco_analyzer.cpp @@ -16,6 +16,7 @@ #include #include #include +#include #include #include #include @@ -46,58 +47,71 @@ analyze(const std::uint32_t n_threads, for (const auto file_id : std::views::iota(0, n_files)) tq.push(file_id, std::monostate{}); + std::exception_ptr eptr; + { // scope to join jthreads std::vector n_tasks(n_files); std::vector workers; for (const auto th_id : std::views::iota(0u, n_threads)) workers.emplace_back([&, n_threads, th_id] { - auto &res = results[th_id]; - for (const auto [file_id, info] : falco::views::enumerate(infos)) - res[file_id].init(mode, info, dups_init[file_id]); - while (true) { - auto tq_lock = tq.wait_and_acquire_lock(); - if (tq.is_finished()) - return; - auto [file_id, task] = tq.pop_and_release_lock(tq_lock); - if (std::holds_alternative(task)) - process_reads(res[file_id], std::get(task)); - else if (std::holds_alternative(task)) - process_reads(res[file_id], std::get(task)); - else if (std::holds_alternative(task)) - process_reads(res[file_id], std::get(task)); - else if (std::holds_alternative(task)) - decompress(std::get(task)); - else { // monostate means read more data - if (!is_active(reads_files[file_id])) { - assert(n_tasks[file_id] == 0); - reset(reads_files[file_id]); - if (n_active_files.fetch_sub(1, std::memory_order_relaxed) == 1) { - // ADS: it would not be wrong it two different threads arrived - // here, but using fetch_sub should prevent that anyway. - tq.request_shutdown(); - return; + try { + auto &res = results[th_id]; + for (const auto [file_id, info] : falco::views::enumerate(infos)) + res[file_id].init(mode, info, dups_init[file_id]); + while (true) { + auto tq_lock = tq.wait_and_acquire_lock(); + if (tq.is_finished()) + return; + auto [file_id, task] = tq.pop_and_release_lock(tq_lock); + if (std::holds_alternative(task)) + process_reads(res[file_id], std::get(task)); + else if (std::holds_alternative(task)) + process_reads(res[file_id], std::get(task)); + else if (std::holds_alternative(task)) + process_reads(res[file_id], std::get(task)); + else if (std::holds_alternative(task)) + decompress(std::get(task)); + else { // monostate means read more data + if (!is_active(reads_files[file_id])) { + assert(n_tasks[file_id] == 0); + reset(reads_files[file_id]); + if (n_active_files.fetch_sub(1, std::memory_order_relaxed) == + 1) { + // ADS: it would not be wrong it two different threads arrived + // here, but using fetch_sub should prevent that anyway. + tq.request_shutdown(); + return; + } + // Must 'continue' here to avoid decrementing n_tasks[file_id] + // below, since within this branch we might exit before + // decrementing to remove the value that we would have added; + // the same is not true of the 'else' case below when make_tasks + // is called. So the ++n_tasks[file_id] is inside that function + // in each case of typeof(reads_files). Unfortunately changes to + // n_tasks need to happen in different places. + continue; } - // Must 'continue' here to avoid decrementing n_tasks[file_id] - // below, since within this branch we might exit before - // decrementing to remove the value that we would have added; the - // same is not true of the 'else' case below when make_tasks is - // called. So the ++n_tasks[file_id] is inside that function in - // each case of typeof(reads_files). Unfortunately changes to - // n_tasks need to happen in different places. - continue; + else + make_tasks(reads_files[file_id], n_threads, file_id, tq, + n_tasks[file_id]); + } + if (n_tasks[file_id].fetch_sub(1, std::memory_order_relaxed) == 1) { + assert(n_tasks[file_id] == 0); + tq.push(file_id, std::monostate{}); } - else - make_tasks(reads_files[file_id], n_threads, file_id, tq, - n_tasks[file_id]); - } - if (n_tasks[file_id].fetch_sub(1, std::memory_order_relaxed) == 1) { - assert(n_tasks[file_id] == 0); - tq.push(file_id, std::monostate{}); } } + catch (...) { + eptr = std::current_exception(); + tq.request_shutdown(); // do this or deadlock + return; + } }); } + if (eptr) + std::rethrow_exception(eptr); + std::vector finalized; finalized.reserve(n_files); for (const auto file_id : std::views::iota(0, n_files)) {