detectors/aare
25
0
Fork 0
mirror of https://github.com/slsdetectorgroup/aare.git synced 2025-06-20 02:57:13 +02:00
Code Issues Actions Packages Releases Activity

MultiThreaded Cluster finder

Browse Source
This commit is contained in:
froejdh_e
2025-01-09 16:53:22 +01:00
parent dc9e10016d
commit cc95561eda
16 changed files with 268 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
include/aare/ClusterFile.hpp
View File

@ -59,10 +59,9 @@ class ClusterFile {
ClusterFile(const std::filesystem::path &fname, size_t chunk_size = 1000,
const std::string &mode = "r");
~ClusterFile();
std::vector<Cluster3x3> read_clusters(size_t n_clusters);
std::vector<Cluster3x3> read_frame(int32_t &out_fnum);
void write_frame(int32_t frame_number,
const ClusterVector<int32_t> &clusters);
ClusterVector<int32_t> read_clusters(size_t n_clusters);
ClusterVector<int32_t> read_frame();
void write_frame(const ClusterVector<int32_t> &clusters);
std::vector<Cluster3x3>
read_cluster_with_cut(size_t n_clusters, double *noise_map, int nx, int ny);

56
include/aare/ClusterFileSink.hpp Normal file
View File

@ -0,0 +1,56 @@
#pragma once
#include <atomic>
#include <filesystem>
#include <thread>
#include "aare/ProducerConsumerQueue.hpp"
#include "aare/ClusterVector.hpp"
#include "aare/ClusterFinderMT.hpp"
namespace aare{
class ClusterFileSink{
ProducerConsumerQueue<ClusterVector<int>>* m_source;
std::atomic<bool> m_stop_requested{false};
std::atomic<bool> m_stopped{true};
std::chrono::milliseconds m_default_wait{1};
std::thread m_thread;
std::ofstream m_file;
void process(){
m_stopped = false;
fmt::print("ClusterFileSink started\n");
while (!m_stop_requested || !m_source->isEmpty()) {
if (ClusterVector<int> *clusters = m_source->frontPtr();
clusters != nullptr) {
// Write clusters to file
int32_t frame_number = clusters->frame_number(); //TODO! Should we store frame number already as int?
uint32_t num_clusters = clusters->size();
m_file.write(reinterpret_cast<const char*>(&frame_number), sizeof(frame_number));
m_file.write(reinterpret_cast<const char*>(&num_clusters), sizeof(num_clusters));
m_file.write(reinterpret_cast<const char*>(clusters->data()), clusters->size() * clusters->item_size());
m_source->popFront();
}else{
std::this_thread::sleep_for(m_default_wait);
}
}
fmt::print("ClusterFileSink stopped\n");
m_stopped = true;
}
public:
ClusterFileSink(ClusterFinderMT<uint16_t, double, int32_t>* source, const std::filesystem::path& fname){
m_source = source->sink();
m_thread = std::thread(&ClusterFileSink::process, this);
m_file.open(fname, std::ios::binary);
}
void stop(){
m_stop_requested = true;
m_thread.join();
m_file.close();
}
};
} // namespace aare

4
include/aare/ClusterFinder.hpp
View File

@ -64,13 +64,13 @@ class ClusterFinder {
m_clusters = ClusterVector<CT>(m_cluster_sizeX, m_cluster_sizeY);
return tmp;
}
void find_clusters(NDView<FRAME_TYPE, 2> frame) {
void find_clusters(NDView<FRAME_TYPE, 2> frame, uint64_t frame_number = 0) {
// // TODO! deal with even size clusters
// // currently 3,3 -> +/- 1
// // 4,4 -> +/- 2
int dy = m_cluster_sizeY / 2;
int dx = m_cluster_sizeX / 2;
m_clusters.set_frame_number(frame_number);
std::vector<CT> cluster_data(m_cluster_sizeX * m_cluster_sizeY);
for (int iy = 0; iy < frame.shape(0); iy++) {
for (int ix = 0; ix < frame.shape(1); ix++) {

101
include/aare/ClusterFinderMT.hpp
View File

@ -7,6 +7,7 @@
#include "aare/NDArray.hpp"
#include "aare/ProducerConsumerQueue.hpp"
#include "aare/ClusterFinder.hpp"
namespace aare {
@ -21,7 +22,6 @@ struct FrameWrapper {
NDArray<uint16_t, 2> data;
};
template <typename FRAME_TYPE = uint16_t, typename PEDESTAL_TYPE = double,
typename CT = int32_t>
class ClusterFinderMT {
@ -32,10 +32,8 @@ class ClusterFinderMT {
using OutputQueue = ProducerConsumerQueue<ClusterVector<int>>;
std::vector<std::unique_ptr<InputQueue>> m_input_queues;
std::vector<std::unique_ptr<OutputQueue>> m_output_queues;
OutputQueue m_sink{1000}; //All clusters go into this queue
OutputQueue m_sink{1000}; // All clusters go into this queue
std::vector<std::unique_ptr<Finder>> m_cluster_finders;
std::vector<std::thread> m_threads;
@ -43,26 +41,24 @@ class ClusterFinderMT {
std::chrono::milliseconds m_default_wait{1};
std::atomic<bool> m_stop_requested{false};
std::atomic<bool> m_processing_threads_stopped{false};
std::atomic<bool> m_processing_threads_stopped{true};
void process(int thread_id) {
auto cf = m_cluster_finders[thread_id].get();
auto q = m_input_queues[thread_id].get();
// TODO! Avoid indexing into the vector every time
fmt::print("Thread {} started\n", thread_id);
//TODO! is this check enough to make sure we process all the frames?
while (!m_stop_requested || !q->isEmpty()) {
if (FrameWrapper *frame = q->frontPtr();
frame != nullptr) {
// TODO! is this check enough to make sure we process all the frames?
while (!m_stop_requested || !q->isEmpty()) {
if (FrameWrapper *frame = q->frontPtr(); frame != nullptr) {
// fmt::print("Thread {} got frame {}, type: {}\n", thread_id,
// frame->frame_number, static_cast<int>(frame->type));
// frame->frame_number, static_cast<int>(frame->type));
switch (frame->type) {
case FrameType::DATA:
cf->find_clusters(
frame->data.view());
cf->find_clusters(frame->data.view(), frame->frame_number);
m_output_queues[thread_id]->write(cf->steal_clusters());
break;
case FrameType::PEDESTAL:
@ -79,22 +75,22 @@ class ClusterFinderMT {
} else {
std::this_thread::sleep_for(m_default_wait);
}
}
fmt::print("Thread {} stopped\n", thread_id);
}
/**
* @brief Collect all the clusters from the output queues and write them to the sink
* @brief Collect all the clusters from the output queues and write them to
* the sink
*/
void collect(){
void collect() {
bool empty = true;
while(!m_stop_requested || !empty || !m_processing_threads_stopped){
while (!m_stop_requested || !empty || !m_processing_threads_stopped) {
empty = true;
for (auto &queue : m_output_queues) {
if (!queue->isEmpty()) {
while(!m_sink.write(std::move(*queue->frontPtr()))){
while (!m_sink.write(std::move(*queue->frontPtr()))) {
std::this_thread::sleep_for(m_default_wait);
}
queue->popFront();
@ -118,7 +114,6 @@ class ClusterFinderMT {
for (size_t i = 0; i < n_threads; i++) {
m_input_queues.emplace_back(std::make_unique<InputQueue>(200));
m_output_queues.emplace_back(std::make_unique<OutputQueue>(200));
}
start();
@ -126,14 +121,22 @@ class ClusterFinderMT {
ProducerConsumerQueue<ClusterVector<int>> *sink() { return &m_sink; }
void start(){
/**
* @brief Start all threads
*/
void start() {
for (size_t i = 0; i < m_n_threads; i++) {
m_threads.push_back(
std::thread(&ClusterFinderMT::process, this, i));
}
m_processing_threads_stopped = false;
m_collect_thread = std::thread(&ClusterFinderMT::collect, this);
}
/**
* @brief Stop all threads
*/
void stop() {
m_stop_requested = true;
for (auto &thread : m_threads) {
@ -143,42 +146,74 @@ class ClusterFinderMT {
m_collect_thread.join();
}
void sync(){
/**
* @brief Wait for all the queues to be empty
*/
void sync() {
for (auto &q : m_input_queues) {
while(!q->isEmpty()){
while (!q->isEmpty()) {
std::this_thread::sleep_for(m_default_wait);
}
}
for (auto &q : m_output_queues) {
while (!q->isEmpty()) {
std::this_thread::sleep_for(m_default_wait);
}
}
while (!m_sink.isEmpty()) {
std::this_thread::sleep_for(m_default_wait);
}
}
/**
* @brief Push a pedestal frame to all the cluster finders. The frames is
* expected to be dark. No photon finding is done. Just pedestal update.
*/
void push_pedestal_frame(NDView<FRAME_TYPE, 2> frame) {
FrameWrapper fw{FrameType::PEDESTAL, 0, NDArray(frame)};
FrameWrapper fw{FrameType::PEDESTAL, 0,
NDArray(frame)}; // TODO! copies the data!
for (auto &queue : m_input_queues) {
while (!queue->write(fw)) {
// fmt::print("push_pedestal_frame: queue full\n");
std::this_thread::sleep_for(m_default_wait);
}
}
}
void find_clusters(NDView<FRAME_TYPE, 2> frame) {
FrameWrapper fw{FrameType::DATA, 0, NDArray(frame)};
while (!m_input_queues[m_current_thread%m_n_threads]->write(fw)) {
/**
* @brief Push the frame to the queue of the next available thread. Function
* returns once the frame is in a queue.
* @note Spin locks with a default wait if the queue is full.
*/
void find_clusters(NDView<FRAME_TYPE, 2> frame, uint64_t frame_number = 0) {
FrameWrapper fw{FrameType::DATA, frame_number,
NDArray(frame)}; // TODO! copies the data!
while (!m_input_queues[m_current_thread % m_n_threads]->write(fw)) {
std::this_thread::sleep_for(m_default_wait);
}
m_current_thread++;
}
ClusterVector<int> steal_clusters(bool realloc_same_capacity = false) {
ClusterVector<int> clusters(3,3);
for (auto &finder : m_cluster_finders) {
clusters += finder->steal_clusters();
auto pedestal() {
if (m_cluster_finders.empty()) {
throw std::runtime_error("No cluster finders available");
}
return clusters;
if(!m_processing_threads_stopped){
throw std::runtime_error("ClusterFinderMT is still running");
}
return m_cluster_finders[0]->pedestal();
}
auto noise() {
if (m_cluster_finders.empty()) {
throw std::runtime_error("No cluster finders available");
}
if(!m_processing_threads_stopped){
throw std::runtime_error("ClusterFinderMT is still running");
}
return m_cluster_finders[0]->noise();
}
// void push(FrameWrapper&& frame) {
// //TODO! need to loop until we are successful
// auto rc = m_input_queue.write(std::move(frame));

20
include/aare/ClusterVector.hpp
View File

@ -41,9 +41,9 @@ template <typename T, typename CoordType=int16_t> class ClusterVector {
* @param capacity initial capacity of the buffer in number of clusters
*/
ClusterVector(size_t cluster_size_x = 3, size_t cluster_size_y = 3,
size_t capacity = 1024)
size_t capacity = 1024, uint64_t frame_number = 0)
: m_cluster_size_x(cluster_size_x), m_cluster_size_y(cluster_size_y),
m_capacity(capacity) {
m_capacity(capacity), m_frame_number(frame_number) {
allocate_buffer(capacity);
}
~ClusterVector() {
@ -55,7 +55,7 @@ template <typename T, typename CoordType=int16_t> class ClusterVector {
ClusterVector(ClusterVector &&other) noexcept
: m_cluster_size_x(other.m_cluster_size_x),
m_cluster_size_y(other.m_cluster_size_y), m_data(other.m_data),
m_size(other.m_size), m_capacity(other.m_capacity) {
m_size(other.m_size), m_capacity(other.m_capacity), m_frame_number(other.m_frame_number) {
other.m_data = nullptr;
other.m_size = 0;
other.m_capacity = 0;
@ -70,9 +70,11 @@ template <typename T, typename CoordType=int16_t> class ClusterVector {
m_data = other.m_data;
m_size = other.m_size;
m_capacity = other.m_capacity;
m_frame_number = other.m_frame_number;
other.m_data = nullptr;
other.m_size = 0;
other.m_capacity = 0;
other.m_frame_number = 0;
}
return *this;
}
@ -147,6 +149,12 @@ template <typename T, typename CoordType=int16_t> class ClusterVector {
return 2*sizeof(CoordType) +
m_cluster_size_x * m_cluster_size_y * sizeof(T);
}
/**
* @brief Return the size in bytes of a single cluster
*/
size_t item_size() const { return element_offset(); }
/**
* @brief Return the offset in bytes for the i-th cluster
*/
@ -176,6 +184,12 @@ template <typename T, typename CoordType=int16_t> class ClusterVector {
uint64_t frame_number() const { return m_frame_number; }
void set_frame_number(uint64_t frame_number) { m_frame_number = frame_number; }
void resize(size_t new_size) {
if (new_size > m_capacity) {
allocate_buffer(new_size);
}
m_size = new_size;
}
private:
void allocate_buffer(size_t new_capacity) {

2
include/aare/File.hpp
View File

@ -36,6 +36,8 @@ class File {
File(File &&other) noexcept;
File& operator=(File &&other) noexcept;
~File() = default;
// void close(); //!< close the file
Frame read_frame(); //!< read one frame from the file at the current position
Frame read_frame(size_t frame_index); //!< read one frame at the position given by frame number