From 801a50830ac0b7f9e3a898b9fbeb2e502d1cac0d Mon Sep 17 00:00:00 2001 From: leonarski_f Date: Thu, 16 Apr 2026 22:35:32 +0200 Subject: [PATCH] IndexerThreadPool: New version - requests are dropped, if no available workers (good for real-time, not good for offline processing...need smarter fix) --- image_analysis/indexing/IndexerThreadPool.cpp | 272 ++++++++++-------- image_analysis/indexing/IndexerThreadPool.h | 41 +-- receiver/JFJochReceiverService.cpp | 2 +- 3 files changed, 181 insertions(+), 134 deletions(-) diff --git a/image_analysis/indexing/IndexerThreadPool.cpp b/image_analysis/indexing/IndexerThreadPool.cpp index 668ae067..3d57769f 100644 --- a/image_analysis/indexing/IndexerThreadPool.cpp +++ b/image_analysis/indexing/IndexerThreadPool.cpp @@ -14,76 +14,24 @@ #include "FFTIndexerCPU.h" #endif -IndexerThreadPool::IndexerThreadPool(const IndexingSettings &settings, const NUMAHWPolicy &numa_policy) - : workers_ready(settings.GetIndexingThreads()), - viable_cell_min_spots(settings.GetViableCellMinSpots()), - stop(false) { - for (size_t i = 0; i < settings.GetIndexingThreads(); ++i) - workers.emplace_back([this, i, numa_policy, settings] { Worker(i, numa_policy, settings); }); - workers_ready.wait(); - - if (failed_start) { - { - std::unique_lock lock(m); - stop = true; - } - cond.notify_all(); - - for (std::thread &worker: workers) { - if (worker.joinable()) - worker.join(); - } - throw JFJochException(JFJochExceptionCategory::GPUCUDAError, - "Cannot configure indexer (likely too many threads, not enough memory)"); +IndexerThread::IndexerThread(const IndexingSettings &settings, int threadid) { + std::unique_lock lock(m); + state = TaskState::STARTING; + worker_thread = std::thread(&IndexerThread::Worker, this, std::cref(settings), threadid); + c_running.wait(lock, [this] { return state != TaskState::STARTING; }); + if (state == TaskState::ERROR) { + worker_thread.join(); + throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, + "Indexer thread initialization failed"); } } -IndexerThreadPool::~IndexerThreadPool() { { - std::unique_lock lock(m); - stop = true; - } - cond.notify_all(); - - for (std::thread &worker: workers) { - if (worker.joinable()) - worker.join(); - } -} - -IndexerResult IndexerThreadPool::Run(const DiffractionExperiment &experiment, const std::vector &recip) { - if (experiment.GetIndexingAlgorithm() == IndexingAlgorithmEnum::None) - return IndexerResult{.lattice = {}, .indexing_time_s = 0}; - if (recip.size() < viable_cell_min_spots) - return IndexerResult{.lattice = {}, .indexing_time_s = 0}; - - // Create a promise/future pair - auto promise = std::make_shared >(); - std::future result = promise->get_future(); { - std::unique_lock lock(m); - - // Don't allow enqueueing after stopping the pool - if (stop) { - throw std::runtime_error("Cannot enqueue on stopped thread pool"); - } - - // Create a task package with the data message and coordinates - taskQueue.emplace(TaskPackage{promise, &experiment, &recip}); - } - - cond.notify_one(); - return result.get(); -} - -void IndexerThreadPool::Worker(int32_t threadIndex, const NUMAHWPolicy &numa_policy, const IndexingSettings &settings) { +void IndexerThread::Worker(const IndexingSettings &settings, int threadid) { try { #ifdef JFJOCH_USE_CUDA auto gpu_count = get_gpu_count(); if (gpu_count > 0) - NUMAHWPolicy::SelectGPUAndItsNUMA(threadIndex % gpu_count); - else - numa_policy.Bind(threadIndex); -#else - numa_policy.Bind(threadIndex); + NUMAHWPolicy::SelectGPUAndItsNUMA(threadid % gpu_count); #endif } catch (const std::exception &e) { spdlog::error("Failed to bind thread to NUMA node: {}", e.what()); @@ -106,10 +54,20 @@ void IndexerThreadPool::Worker(int32_t threadIndex, const NUMAHWPolicy &numa_pol } } catch (const std::exception &e) { spdlog::error("Failed to initialize GPU indexer: {}", e.what()); - failed_start = true; + { + std::unique_lock lock(m); + state = TaskState::ERROR; + } + c_running.notify_all(); + return; } catch (...) { spdlog::error("Failed to initialize GPU indexer"); - failed_start = true; + { + std::unique_lock lock(m); + state = TaskState::ERROR; + } + c_running.notify_all(); + return; } #endif #ifdef JFJOCH_USE_FFTW @@ -119,65 +77,145 @@ void IndexerThreadPool::Worker(int32_t threadIndex, const NUMAHWPolicy &numa_pol fftw_indexer = std::make_unique(settings); } catch (const std::exception &e) { spdlog::error("Failed to initialize FFTW indexer: {}", e.what()); - failed_start = true; - } catch (...) { - spdlog::error("Failed to initialize FFTW indexer"); - failed_start = true; - } -#endif - - workers_ready.count_down(); - - while (true) { - TaskPackage task; { std::unique_lock lock(m); - - // Add a timeout to the wait to ensure we can exit even if no notification - cond.wait_for(lock, std::chrono::milliseconds(50), [this] { - return stop || !taskQueue.empty(); - }); - - // Check for exit conditions - if (stop && taskQueue.empty()) - return; // Exit cleanly - - if (!taskQueue.empty()) { - task = std::move(taskQueue.front()); - taskQueue.pop(); - } else { - continue; // No tasks, go back to waiting - } + state = TaskState::ERROR; } - try { - IndexerResult result; + c_running.notify_all(); + return; + } catch (...) { + spdlog::error("Failed to initialize FFTW indexer"); + { + std::unique_lock lock(m); + state = TaskState::ERROR; + } + c_running.notify_all(); + return; + } +#endif + { + std::unique_lock lock(m); + state = TaskState::IDLE; + } + c_running.notify_all(); - auto algorithm = task.experiment->GetIndexingAlgorithm(); - Indexer *indexer = nullptr; + while (true) { + std::unique_ptr input; + // Look for task + handle stop + { + std::unique_lock lock(m); + c_start.wait(lock, [this] { return stop || state == TaskState::READY; }); + if (stop) + return; + input = std::move(task_input); + } + if (input) { + std::unique_ptr tmp_result; + try { + auto algorithm = input->experiment.GetIndexingAlgorithm(); + Indexer *indexer = nullptr; - if (algorithm == IndexingAlgorithmEnum::FFT && fft_indexer) { - indexer = fft_indexer.get(); - } else if (algorithm == IndexingAlgorithmEnum::FFBIDX && ffbidx_indexer) { - indexer = ffbidx_indexer.get(); - } else if (algorithm == IndexingAlgorithmEnum::FFTW && fftw_indexer) { - indexer = fftw_indexer.get(); - } else if (algorithm == IndexingAlgorithmEnum::Auto) { - throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, - "Internal error: Invalid indexing algorithm provided"); + if (algorithm == IndexingAlgorithmEnum::FFT && fft_indexer) { + indexer = fft_indexer.get(); + } else if (algorithm == IndexingAlgorithmEnum::FFBIDX && ffbidx_indexer) { + indexer = ffbidx_indexer.get(); + } else if (algorithm == IndexingAlgorithmEnum::FFTW && fftw_indexer) { + indexer = fftw_indexer.get(); + } else if (algorithm == IndexingAlgorithmEnum::Auto) { + throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, + "Internal error: Invalid indexing algorithm provided"); + } + + if (indexer) { + indexer->Setup(input->experiment); + tmp_result = std::make_unique(indexer->Run(input->recip)); + } + } catch (std::exception &e) { + tmp_result = nullptr; + spdlog::error("Indexer thread {} failed: {}", threadid, e.what()); } - - if (indexer) { - indexer->Setup(*task.experiment); - result = indexer->Run(*task.recip); + { + std::unique_lock lock(m); + state = TaskState::COMPLETED; + result = std::move(tmp_result); + c_done.notify_all(); } - - // Set the result via the promise - if (task.promise) { - task.promise->set_value(result); - } - } catch (std::exception &e) { - if (task.promise) - task.promise->set_exception(std::current_exception()); } } } + +void IndexerThread::Finalize() { + { + std::unique_lock lock(m); + stop = true; + } + c_start.notify_one(); + if (worker_thread.joinable()) + worker_thread.join(); +} + +std::unique_ptr IndexerThread::Run(const DiffractionExperiment &experiment, + const std::vector &recip) { + std::unique_ptr tmp_result; + { + std::unique_lock lock(m); + if (stop) + return nullptr; + if (state != TaskState::IDLE) + return nullptr; + task_input = std::make_unique(std::cref(experiment), std::cref(recip)); + state = TaskState::READY; + } + c_start.notify_one(); + { + std::unique_lock lock(m); + c_done.wait(lock, [this] { return state == TaskState::COMPLETED; }); + tmp_result = std::move(result); + state = TaskState::IDLE; + } + return std::move(tmp_result); +} + +IndexerThread::~IndexerThread() { + Finalize(); +} + +IndexerThreadPool::IndexerThreadPool(const IndexingSettings &settings) + : worker_busy(settings.GetIndexingThreads(), 0), + viable_cell_min_spots(settings.GetViableCellMinSpots()) { + for (size_t i = 0; i < settings.GetIndexingThreads(); ++i) + tasks.emplace_back(std::make_unique(std::cref(settings), i)); +} + + +IndexerResult IndexerThreadPool::Run(const DiffractionExperiment &experiment, const std::vector &recip) { + if (experiment.GetIndexingAlgorithm() == IndexingAlgorithmEnum::None) + return IndexerResult{.lattice = {}, .indexing_time_s = 0}; + if (recip.size() < viable_cell_min_spots) + return IndexerResult{.lattice = {}, .indexing_time_s = 0}; + + // Check if there is available worker + int task = -1; + { + std::unique_lock lock(m); + for (int i = 0; i < tasks.size(); i++) { + if (worker_busy[i] == 0) { + task = i; + worker_busy[i] = 1; + break; + } + } + } + + std::unique_ptr result; + if (task >= 0) { + result = tasks[task]->Run(experiment, recip); + { + std::unique_lock lock(m); + worker_busy[task] = 0; + } + } + if (result) + return *result; + return IndexerResult{.lattice = {}, .indexing_time_s = 0}; +} diff --git a/image_analysis/indexing/IndexerThreadPool.h b/image_analysis/indexing/IndexerThreadPool.h index f4c42d75..efccc369 100644 --- a/image_analysis/indexing/IndexerThreadPool.h +++ b/image_analysis/indexing/IndexerThreadPool.h @@ -21,31 +21,40 @@ #include "../common/NUMAHWPolicy.h" #include "Indexer.h" -class IndexerThreadPool { - std::atomic failed_start = false; - struct TaskPackage { - std::shared_ptr> promise; - const DiffractionExperiment *experiment; - const std::vector *recip; +class IndexerThread { + struct TaskInput { + const DiffractionExperiment &experiment; + const std::vector &recip; }; - std::vector workers; - + bool stop = false; + enum class TaskState {STARTING, IDLE, READY, COMPLETED, ERROR} state = TaskState::STARTING; std::mutex m; - std::condition_variable cond; - std::queue taskQueue; - std::latch workers_ready; + std::condition_variable c_running; + std::condition_variable c_start; + std::condition_variable c_done; + std::unique_ptr result = nullptr; + std::unique_ptr task_input = nullptr; + std::thread worker_thread; + void Worker(const IndexingSettings& settings, int threadid); +public: + IndexerThread(const IndexingSettings& settings, int threadid); + ~IndexerThread(); + std::unique_ptr Run(const DiffractionExperiment &experiment, const std::vector &recip); + void Finalize(); +}; + +class IndexerThreadPool { + std::mutex m; + std::vector worker_busy; + std::vector> tasks; const int64_t viable_cell_min_spots; - bool stop; - void Worker(int32_t threadIndex, const NUMAHWPolicy &numa_policy, const IndexingSettings& settings); public: - IndexerThreadPool(const IndexingSettings& settings, const NUMAHWPolicy &numa_policy = NUMAHWPolicy()); - ~IndexerThreadPool(); - + IndexerThreadPool(const IndexingSettings& settings); IndexerResult Run(const DiffractionExperiment& experiment, const std::vector& recip); }; diff --git a/receiver/JFJochReceiverService.cpp b/receiver/JFJochReceiverService.cpp index 79151fae..e44655f3 100644 --- a/receiver/JFJochReceiverService.cpp +++ b/receiver/JFJochReceiverService.cpp @@ -329,7 +329,7 @@ JFJochReceiverService &JFJochReceiverService::Indexing(const IndexingSettings &i if (input.GetAlgorithm() != IndexingAlgorithmEnum::None) { logger.Info("Creating indexing thread pool..."); - indexer_thread_pool = std::make_unique(input, numa_policy); + indexer_thread_pool = std::make_unique(input); logger.Info(" ... done"); } return *this;