detectors/aare
25
0
Fork 0
mirror of https://github.com/slsdetectorgroup/aare.git synced 2026-08-05 22:02:24 +02:00
Code Issues Actions Packages Releases Activity

using circular fifo

Browse Source
This commit is contained in:
Erik Frojdh
2026-08-04 16:26:27 +02:00
parent 07b994c72e
commit ff3c088bfd
9 changed files with 325 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/aare/Backoff.hpp
+48
View File
@@ -0,0 +1,48 @@
// SPDX-License-Identifier: MPL-2.0
#pragma once
#include <chrono>
#include <thread>
namespace aare {
/**
* @brief Hint the CPU that this is a spin-wait. Prefer this over yield() when
* the wait is expected to be short.
*/
inline void cpu_relax() noexcept {
#if defined(__x86_64__) || defined(__i386__)
__builtin_ia32_pause();
#elif defined(__aarch64__)
asm volatile("yield" ::: "memory");
#else
std::this_thread::yield();
#endif
}
/**
* @brief Escalating wait for producer/consumer idle and backpressure loops.
*
* Starts with a pause instruction (sub-microsecond, no syscall), then yield,
* then a short sleep. Call reset() whenever work arrives so a busy pipeline
* never leaves the pause tier.
*/
class Backoff {
int m_count{0};
public:
void reset() noexcept { m_count = 0; }
void pause() noexcept {
if (m_count < 64) {
cpu_relax();
} else if (m_count < 256) {
std::this_thread::yield();
} else {
std::this_thread::sleep_for(std::chrono::microseconds(50));
}
++m_count;
}
};
} // namespace aare
include/aare/CircularFifo.hpp
+42 -2
View File
@@ -1,11 +1,13 @@
// SPDX-License-Identifier: MPL-2.0
#pragma once
#include <cassert>
#include <chrono>
#include <fmt/color.h>
#include <fmt/format.h>
#include <memory>
#include <thread>
#include <utility>
#include "aare/ProducerConsumerQueue.hpp"
@@ -29,6 +31,23 @@ template <class ItemType> class CircularFifo {
}
}
/**
* @brief Construct a fifo and seed the free list using a factory.
* @param size number of items circulating in the fifo
* @param make_item callable invoked as make_item(i) for each slot index i
*
* Use this instead of CircularFifo(size) when the items need to be
* initialized, for example to hold a preallocated buffer or to carry
* their own slot index.
*/
template <class F>
CircularFifo(uint32_t size, F make_item)
: fifo_size(size), free_slots(size + 1), filled_slots(size + 1) {
for (size_t i = 0; i < fifo_size; ++i) {
free_slots.write(make_item(i));
}
}
bool next() {
// TODO! avoid default constructing ItemType
ItemType it;
@@ -47,6 +66,13 @@ template <class ItemType> class CircularFifo {
auto numFreeSlots() const noexcept { return free_slots.sizeGuess(); }
auto isFull() const noexcept { return filled_slots.isFull(); }
/**
* @brief True if there are no filled slots waiting to be consumed.
* @note Prefer this over numFilledSlots() == 0 since sizeGuess() may
* under-report when called from the producing thread.
*/
auto isEmpty() const noexcept { return filled_slots.isEmpty(); }
ItemType pop_free() {
ItemType v;
while (!free_slots.read(v))
@@ -75,8 +101,22 @@ template <class ItemType> class CircularFifo {
ItemType *frontPtr() { return filled_slots.frontPtr(); }
// TODO! Add function to move item from filled to free to be used
// with the frontPtr function
/**
* @brief Return the front filled item to the free list. To be used
* together with frontPtr() once the item has been consumed in place.
* @warning The fifo must not be empty when calling this.
*
* The item is written to the free list before it is popped from the
* filled list, so it can never be dropped. The write cannot fail: both
* queues hold size + 1 slots while only size items circulate.
*/
void recycle_front() {
ItemType *it = filled_slots.frontPtr();
assert(it != nullptr);
[[maybe_unused]] const bool ok = free_slots.write(std::move(*it));
assert(ok);
filled_slots.popFront();
}
template <class... Args> void push_value(Args &&...recordArgs) {
while (!filled_slots.write(std::forward<Args>(recordArgs)...))
include/aare/ClusterCollector.hpp
+7 -4
View File
@@ -2,7 +2,9 @@
#pragma once
#include <atomic>
#include <thread>
#include <vector>
#include "aare/Backoff.hpp"
#include "aare/ClusterFinderMT.hpp"
#include "aare/ClusterVector.hpp"
#include "aare/ProducerConsumerQueue.hpp"
@@ -16,20 +18,21 @@ class ClusterCollector {
ProducerConsumerQueue<ClusterVector<ClusterType>> *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::vector<ClusterVector<ClusterType>> m_clusters;
void process() {
m_stopped = false;
fmt::print("ClusterCollector started\n");
while (!m_stop_requested || !m_source->isEmpty()) {
Backoff backoff;
while (!m_stop_requested) {
if (ClusterVector<ClusterType> *clusters = m_source->frontPtr();
clusters != nullptr) {
backoff.reset();
m_clusters.push_back(std::move(*clusters));
m_source->popFront();
} else {
std::this_thread::sleep_for(m_default_wait);
backoff.pause();
}
}
fmt::print("ClusterCollector stopped\n");
@@ -57,4 +60,4 @@ class ClusterCollector {
}
};
} // namespace aare
} // namespace aare
include/aare/ClusterFinder.hpp
+7
View File
@@ -37,6 +37,8 @@ class ClusterFinder {
NDArray<PEDESTAL_TYPE, 2> m_threshold;
NDArray<PEDESTAL_TYPE, 2> m_pd_corrected_frame;
double all=0;
public:
/**
* @brief Construct a new ClusterFinder object
@@ -104,6 +106,7 @@ class ClusterFinder {
template <bool CheckBounds>
void process_pixel(const NDView<FRAME_TYPE, 2> &frame,
const int iy, const int ix) {
constexpr int dy = ClusterSizeY / 2;
constexpr int dx = ClusterSizeX / 2;
constexpr int has_center_pixel_x = ClusterSizeX % 2;
@@ -221,9 +224,11 @@ class ClusterFinder {
public:
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
constexpr int dy = ClusterSizeY / 2;
constexpr int dx = ClusterSizeX / 2;
constexpr int has_center_pixel_x = ClusterSizeX % 2;
@@ -252,6 +257,8 @@ class ClusterFinder {
for (ssize_t i = 0; i < n_pixels; i++) {
corrected[i] = static_cast<PEDESTAL_TYPE>(frame_data[i]) - pd[i];
}
// all += corrected[0];
// return;
// Interior pixels can skip the per-neighbour bounds checks; pixels
// within dx/dy of an edge take the bounds-checked path. Iteration order
include/aare/ClusterFinderMT.hpp
+140 -39
View File
@@ -6,22 +6,74 @@
#include <thread>
#include <vector>
#include "aare/Backoff.hpp"
#include "aare/CircularFifo.hpp"
#include "aare/ClusterFinder.hpp"
#include "aare/NDArray.hpp"
#include "aare/ProducerConsumerQueue.hpp"
#include "aare/logger.hpp"
#include <ctime>
namespace aare {
inline uint64_t thread_cpu_ns() {
timespec ts;
clock_gettime(CLOCK_THREAD_CPUTIME_ID, &ts);
return static_cast<uint64_t>(ts.tv_sec) * 1'000'000'000ULL + ts.tv_nsec;
}
inline uint64_t wall_ns() {
timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return static_cast<uint64_t>(ts.tv_sec) * 1'000'000'000ULL + ts.tv_nsec;
}
struct DualTimer {
uint64_t w0, c0;
DualTimer() : w0(wall_ns()), c0(thread_cpu_ns()) {}
// returns {wall_ns, cpu_ns}
std::pair<uint64_t, uint64_t> elapsed() const {
uint64_t c = thread_cpu_ns() - c0;
return {wall_ns() - w0, c};
}
};
enum class FrameType {
DATA,
PEDESTAL,
};
struct FrameWrapper {
FrameType type;
uint64_t frame_number;
NDArray<uint16_t, 2> data;
/**
* @brief Ticket identifying a frame buffer in a FramePool. Trivially
* copyable, the buffer itself never travels through the queues.
*/
struct FrameRef {
FrameType type{};
uint32_t slot{};
uint64_t frame_number{};
};
/**
* @brief Fixed set of frame buffers allocated once at construction.
*
* Buffers are addressed by slot index and are never moved or reallocated, so
* the queues only need to pass a FrameRef around. This keeps the per frame
* allocation out of the hot path entirely.
*/
class FramePool {
std::vector<NDArray<uint16_t, 2>> m_buffers;
public:
FramePool(size_t depth, Shape<2> shape) {
m_buffers.reserve(depth); // no reallocation, slots stay stable
for (size_t i = 0; i < depth; ++i) {
m_buffers.emplace_back(shape);
}
}
NDArray<uint16_t, 2> &operator[](uint32_t slot) { return m_buffers[slot]; }
const NDArray<uint16_t, 2> &operator[](uint32_t slot) const {
return m_buffers[slot];
}
size_t size() const { return m_buffers.size(); }
};
/**
@@ -42,17 +94,18 @@ class ClusterFinderMT {
size_t m_current_thread{0};
size_t m_n_threads{0};
using Finder = ClusterFinder<ClusterType, FRAME_TYPE, PEDESTAL_TYPE>;
using InputQueue = ProducerConsumerQueue<FrameWrapper>;
using InputQueue = CircularFifo<FrameRef>;
using OutputQueue = ProducerConsumerQueue<ClusterVector<ClusterType>>;
std::vector<std::unique_ptr<InputQueue>> m_input_queues;
std::vector<std::unique_ptr<OutputQueue>> m_output_queues;
std::vector<std::unique_ptr<FramePool>> m_frame_pools;
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;
std::thread m_collect_thread;
std::chrono::milliseconds m_default_wait{1};
std::chrono::microseconds m_default_wait{50};
private:
std::atomic<bool> m_stop_requested{false};
@@ -65,31 +118,35 @@ class ClusterFinderMT {
void process(int thread_id) {
auto cf = m_cluster_finders[thread_id].get();
auto q = m_input_queues[thread_id].get();
bool realloc_same_capacity = true;
auto *pool = m_frame_pools[thread_id].get();
Backoff backoff;
while (!m_stop_requested || !q->isEmpty()) {
if (FrameWrapper *frame = q->frontPtr(); frame != nullptr) {
if (FrameRef *ref = q->frontPtr(); ref != nullptr) {
backoff.reset();
auto view = (*pool)[ref->slot].view();
switch (frame->type) {
case FrameType::DATA:
cf->find_clusters(frame->data.view(), frame->frame_number);
switch (ref->type) {
case FrameType::DATA: {
cf->find_clusters(view, ref->frame_number);
// Steal before the write so a failed write cannot drop the
// clusters by re-stealing an empty vector on retry.
auto clusters = cf->steal_clusters(true);
while (!m_output_queues[thread_id]->write(
cf->steal_clusters(realloc_same_capacity))) {
std::this_thread::sleep_for(m_default_wait);
std::move(clusters))) {
backoff.pause();
}
break;
}
case FrameType::PEDESTAL:
m_cluster_finders[thread_id]->push_pedestal_frame(
frame->data.view());
m_cluster_finders[thread_id]->push_pedestal_frame(view);
break;
}
// frame is processed now discard it
m_input_queues[thread_id]->popFront();
// frame is processed, hand the buffer back to the free list
q->recycle_front();
} else {
std::this_thread::sleep_for(m_default_wait);
backoff.pause();
}
}
}
@@ -100,18 +157,24 @@ class ClusterFinderMT {
*/
void collect() {
bool empty = true;
Backoff backoff;
while (!m_stop_requested || !empty || !m_processing_threads_stopped) {
empty = true;
bool moved_any = false;
for (auto &queue : m_output_queues) {
if (!queue->isEmpty()) {
while (!m_sink.write(std::move(*queue->frontPtr()))) {
std::this_thread::sleep_for(m_default_wait);
while (auto *front = queue->frontPtr()) {
while (!m_sink.write(std::move(*front))) {
backoff.pause();
}
queue->popFront();
empty = false;
moved_any = true;
}
}
empty = !moved_any;
if (moved_any) {
backoff.reset();
} else {
backoff.pause();
}
}
}
@@ -124,16 +187,22 @@ class ClusterFinderMT {
* @param capacity initial capacity of the cluster vector. Should match
* expected number of clusters in a frame per frame.
* @param n_threads number of threads to use
* @param queue_depth number of frame buffers per thread. These are
* allocated once and recycled, so the total resident frame memory is
* n_threads * queue_depth * frame size. Keeping the in flight data below
* the L3 size keeps the per frame copy cheap.
*/
ClusterFinderMT(Shape<2> image_size, PEDESTAL_TYPE nSigma = 5.0,
size_t capacity = 2000, size_t n_threads = 3)
size_t capacity = 2000, size_t n_threads = 3,
size_t queue_depth = 16)
: m_n_threads(n_threads) {
LOG(logDEBUG1) << "ClusterFinderMT: "
<< "image_size: " << image_size[0] << "x"
<< image_size[1] << ", nSigma: " << nSigma
<< ", capacity: " << capacity
<< ", n_threads: " << n_threads;
<< ", n_threads: " << n_threads
<< ", queue_depth: " << queue_depth;
for (size_t i = 0; i < n_threads; i++) {
m_cluster_finders.push_back(
@@ -142,7 +211,13 @@ class ClusterFinderMT {
image_size, nSigma, capacity));
}
for (size_t i = 0; i < n_threads; i++) {
m_input_queues.emplace_back(std::make_unique<InputQueue>(200));
m_frame_pools.emplace_back(
std::make_unique<FramePool>(queue_depth, image_size));
m_input_queues.emplace_back(std::make_unique<InputQueue>(
static_cast<uint32_t>(queue_depth), [](size_t slot) {
return FrameRef{FrameType::DATA,
static_cast<uint32_t>(slot), 0};
}));
m_output_queues.emplace_back(std::make_unique<OutputQueue>(200));
}
// TODO! Should we start automatically?
@@ -212,13 +287,22 @@ class ClusterFinderMT {
* 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)}; // TODO! copies the data!
for (size_t i = 0; i < m_n_threads; ++i) {
auto *q = m_input_queues[i].get();
Backoff backoff;
for (auto &queue : m_input_queues) {
while (!queue->write(fw)) {
std::this_thread::sleep_for(m_default_wait);
FrameRef ref;
while (!q->try_pop_free(ref)) {
backoff.pause();
}
ref.type = FrameType::PEDESTAL;
ref.frame_number = 0;
(*m_frame_pools[i])[ref.slot].copy_from(frame);
// Cannot fail, the free list is what limits how many frames are
// in flight so there is always room in the filled list.
q->try_push_value(ref);
}
}
@@ -228,11 +312,28 @@ class ClusterFinderMT {
* @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);
const size_t tid = m_current_thread % m_n_threads;
auto *q = m_input_queues[tid].get();
Backoff backoff;
FrameRef ref;
while (!q->try_pop_free(ref)) {
backoff.pause();
}
ref.type = FrameType::DATA;
ref.frame_number = frame_number;
// DualTimer dt;
(*m_frame_pools[tid])[ref.slot].copy_from(frame);
// auto [wall_ns, cpu_ns] = dt.elapsed();
// std::cerr << "ClusterFinderMT: find_clusters: copied frame "
// << frame_number << " took " << wall_ns/1000.0 << " wall_us and "
// << cpu_ns/1000.0 << " cpu_us" << std::endl;
// Cannot fail, the free list is what limits how many frames are in
// flight so there is always room in the filled list.
q->try_push_value(ref);
m_current_thread++;
}
include/aare/ClusterVector.hpp
+18 -15
View File
@@ -53,11 +53,14 @@ class ClusterVector<Cluster<T, ClusterSizeX, ClusterSizeY, CoordType>> {
m_data.reserve(capacity);
}
// Move constructor
ClusterVector(ClusterVector &&other) noexcept
: m_data(other.m_data), m_frame_number(other.m_frame_number) {
other.m_data.clear();
}
// // Move constructor
// ClusterVector(ClusterVector &&other) noexcept
// : m_data(other.m_data), m_frame_number(other.m_frame_number) {
// other.m_data.clear();
// }
ClusterVector(ClusterVector &&other) noexcept = default;
ClusterVector &operator=(ClusterVector &&other) noexcept = default;
/**
* @brief Create a copy of the clustervector by filtering clusters in the
@@ -80,16 +83,16 @@ class ClusterVector<Cluster<T, ClusterSizeX, ClusterSizeY, CoordType>> {
return result;
}
// Move assignment operator
ClusterVector &operator=(ClusterVector &&other) noexcept {
if (this != &other) {
m_data = other.m_data;
m_frame_number = other.m_frame_number;
other.m_data.clear();
other.m_frame_number = 0;
}
return *this;
}
// // Move assignment operator
// ClusterVector &operator=(ClusterVector &&other) noexcept {
// if (this != &other) {
// m_data = other.m_data;
// m_frame_number = other.m_frame_number;
// other.m_data.clear();
// other.m_frame_number = 0;
// }
// return *this;
// }
/**
* @brief Sum the pixels in each cluster
include/aare/NDArray.hpp
+17
View File
@@ -146,6 +146,23 @@ class NDArray : public ArrayExpr<NDArray<T, Ndim>, Ndim> {
*/
~NDArray() { delete[] data_; }
/**
* @brief Copy data from a view of matching shape into this array without
* reallocating.
* @param v view to copy from, must have the same shape as this array
* @throws std::runtime_error if the shapes differ
*
* Use this instead of assigning a new NDArray when the buffer needs to be
* kept, for example when the array is part of a preallocated pool.
*/
void copy_from(const NDView<T, Ndim> v) {
if (v.shape() != shape_) {
throw std::runtime_error(LOCATION +
"Shape mismatch in NDArray::copy_from");
}
std::copy(v.begin(), v.end(), begin());
}
///////////////////////////////////////////////////////////////////////////////
// Iterators and indexing
//
python/src/bind_ClusterFinderMT.hpp
+6 -3
View File
@@ -33,9 +33,10 @@ void define_ClusterFinderMT(py::module &m, const std::string &typestr) {
py::class_<ClusterFinderMT<ClusterType, uint16_t, pd_type>>(
m, class_name.c_str())
.def(py::init<Shape<2>, pd_type, size_t, size_t>(),
.def(py::init<Shape<2>, pd_type, size_t, size_t, size_t>(),
py::arg("image_size"), py::arg("n_sigma") = 5.0,
py::arg("capacity") = 2048, py::arg("n_threads") = 3)
py::arg("capacity") = 2048, py::arg("n_threads") = 3,
py::arg("queue_depth") = 16)
.def("push_pedestal_frame",
[](ClusterFinderMT<ClusterType, uint16_t, pd_type> &self,
py::array_t<uint16_t> frame) {
@@ -47,8 +48,10 @@ void define_ClusterFinderMT(py::module &m, const std::string &typestr) {
[](ClusterFinderMT<ClusterType, uint16_t, pd_type> &self,
py::array_t<uint16_t> frame, uint64_t frame_number) {
auto view = make_view_2d(frame);
// Release the GIL for the copy and any free-list wait so a
// Python sink/consumer can keep draining concurrently.
py::gil_scoped_release release;
self.find_clusters(view, frame_number);
return;
},
py::arg(), py::arg("frame_number") = 0)
.def_property_readonly(
src/ClusterFinderMT.test.cpp
+40 -2
View File
@@ -21,14 +21,21 @@ class ClusterFinderMTWrapper
public:
ClusterFinderMTWrapper(Shape<2> image_size, PEDESTAL_TYPE nSigma = 5.0,
size_t capacity = 2000, size_t n_threads = 3)
size_t capacity = 2000, size_t n_threads = 3,
size_t queue_depth = 16)
: ClusterFinderMT<ClusterType, FRAME_TYPE, PEDESTAL_TYPE>(
image_size, nSigma, capacity, n_threads) {}
image_size, nSigma, capacity, n_threads, queue_depth) {}
size_t get_m_input_queues_size() const {
return this->m_input_queues.size();
}
size_t get_m_frame_pools_size() const { return this->m_frame_pools.size(); }
size_t get_frame_pool_depth(size_t thread_index) const {
return this->m_frame_pools[thread_index]->size();
}
size_t get_m_output_queues_size() const {
return this->m_output_queues.size();
}
@@ -100,3 +107,34 @@ TEST_CASE("multithreaded cluster finder", "[.with-data]") {
auto clustervec = clustercollector.steal_clusters();
// CHECK(clustervec.size() == ) //dont know how many clusters to expect
}
TEST_CASE("frame buffers are recycled when pushing more frames than the pool "
"holds",
"[.files]") {
using ClusterType = Cluster<int32_t, 3, 3>;
const size_t n_threads = 2;
const size_t queue_depth = 4;
const size_t n_frames = 5 * queue_depth;
const Shape<2> image_size{10, 10};
ClusterFinderMTWrapper<ClusterType> cf(image_size, 5, 200, n_threads,
queue_depth);
CHECK(cf.get_m_frame_pools_size() == n_threads);
for (size_t i = 0; i < n_threads; ++i) {
CHECK(cf.get_frame_pool_depth(i) == queue_depth);
}
NDArray<uint16_t, 2> frame(image_size, 0);
// More frames than the pool holds, so this only completes if the workers
// return the buffers to the free list.
for (size_t i = 0; i < n_frames; ++i) {
cf.find_clusters(frame.view(), i);
}
cf.stop();
CHECK(cf.m_input_queues_are_empty() == true);
}