From ff3c088bfda1df7db362a19098f079ece199c6f1 Mon Sep 17 00:00:00 2001 From: Erik Frojdh Date: Tue, 4 Aug 2026 16:26:27 +0200 Subject: [PATCH] using circular fifo --- include/aare/Backoff.hpp | 48 ++++++++ include/aare/CircularFifo.hpp | 44 ++++++- include/aare/ClusterCollector.hpp | 11 +- include/aare/ClusterFinder.hpp | 7 ++ include/aare/ClusterFinderMT.hpp | 179 ++++++++++++++++++++++------ include/aare/ClusterVector.hpp | 33 ++--- include/aare/NDArray.hpp | 17 +++ python/src/bind_ClusterFinderMT.hpp | 9 +- src/ClusterFinderMT.test.cpp | 42 ++++++- 9 files changed, 325 insertions(+), 65 deletions(-) create mode 100644 include/aare/Backoff.hpp diff --git a/include/aare/Backoff.hpp b/include/aare/Backoff.hpp new file mode 100644 index 0000000..617c8aa --- /dev/null +++ b/include/aare/Backoff.hpp @@ -0,0 +1,48 @@ +// SPDX-License-Identifier: MPL-2.0 +#pragma once + +#include +#include + +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 diff --git a/include/aare/CircularFifo.hpp b/include/aare/CircularFifo.hpp index e0f4842..fb275b7 100644 --- a/include/aare/CircularFifo.hpp +++ b/include/aare/CircularFifo.hpp @@ -1,11 +1,13 @@ // SPDX-License-Identifier: MPL-2.0 #pragma once +#include #include #include #include #include #include +#include #include "aare/ProducerConsumerQueue.hpp" @@ -29,6 +31,23 @@ template 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 + 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 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 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 void push_value(Args &&...recordArgs) { while (!filled_slots.write(std::forward(recordArgs)...)) diff --git a/include/aare/ClusterCollector.hpp b/include/aare/ClusterCollector.hpp index 23666b7..783b133 100644 --- a/include/aare/ClusterCollector.hpp +++ b/include/aare/ClusterCollector.hpp @@ -2,7 +2,9 @@ #pragma once #include #include +#include +#include "aare/Backoff.hpp" #include "aare/ClusterFinderMT.hpp" #include "aare/ClusterVector.hpp" #include "aare/ProducerConsumerQueue.hpp" @@ -16,20 +18,21 @@ class ClusterCollector { ProducerConsumerQueue> *m_source; std::atomic m_stop_requested{false}; std::atomic m_stopped{true}; - std::chrono::milliseconds m_default_wait{1}; std::thread m_thread; std::vector> 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 *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 \ No newline at end of file +} // namespace aare diff --git a/include/aare/ClusterFinder.hpp b/include/aare/ClusterFinder.hpp index af41cda..e3f8c6b 100644 --- a/include/aare/ClusterFinder.hpp +++ b/include/aare/ClusterFinder.hpp @@ -37,6 +37,8 @@ class ClusterFinder { NDArray m_threshold; NDArray m_pd_corrected_frame; + double all=0; + public: /** * @brief Construct a new ClusterFinder object @@ -104,6 +106,7 @@ class ClusterFinder { template void process_pixel(const NDView &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, 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(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 diff --git a/include/aare/ClusterFinderMT.hpp b/include/aare/ClusterFinderMT.hpp index 2c54bf6..be92e1a 100644 --- a/include/aare/ClusterFinderMT.hpp +++ b/include/aare/ClusterFinderMT.hpp @@ -6,22 +6,74 @@ #include #include +#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 namespace aare { +inline uint64_t thread_cpu_ns() { + timespec ts; + clock_gettime(CLOCK_THREAD_CPUTIME_ID, &ts); + return static_cast(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(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 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 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> 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 &operator[](uint32_t slot) { return m_buffers[slot]; } + const NDArray &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; - using InputQueue = ProducerConsumerQueue; + using InputQueue = CircularFifo; using OutputQueue = ProducerConsumerQueue>; std::vector> m_input_queues; std::vector> m_output_queues; + std::vector> m_frame_pools; OutputQueue m_sink{1000}; // All clusters go into this queue std::vector> m_cluster_finders; std::vector m_threads; std::thread m_collect_thread; - std::chrono::milliseconds m_default_wait{1}; + std::chrono::microseconds m_default_wait{50}; private: std::atomic 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(200)); + m_frame_pools.emplace_back( + std::make_unique(queue_depth, image_size)); + m_input_queues.emplace_back(std::make_unique( + static_cast(queue_depth), [](size_t slot) { + return FrameRef{FrameType::DATA, + static_cast(slot), 0}; + })); m_output_queues.emplace_back(std::make_unique(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) { - 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, 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++; } diff --git a/include/aare/ClusterVector.hpp b/include/aare/ClusterVector.hpp index 0adcb5e..ad49f77 100644 --- a/include/aare/ClusterVector.hpp +++ b/include/aare/ClusterVector.hpp @@ -53,11 +53,14 @@ class ClusterVector> { 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> { 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 diff --git a/include/aare/NDArray.hpp b/include/aare/NDArray.hpp index 079915d..68729dc 100644 --- a/include/aare/NDArray.hpp +++ b/include/aare/NDArray.hpp @@ -146,6 +146,23 @@ class NDArray : public ArrayExpr, 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 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 // diff --git a/python/src/bind_ClusterFinderMT.hpp b/python/src/bind_ClusterFinderMT.hpp index 2484d98..774b3bf 100644 --- a/python/src/bind_ClusterFinderMT.hpp +++ b/python/src/bind_ClusterFinderMT.hpp @@ -33,9 +33,10 @@ void define_ClusterFinderMT(py::module &m, const std::string &typestr) { py::class_>( m, class_name.c_str()) - .def(py::init, pd_type, size_t, size_t>(), + .def(py::init, 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 &self, py::array_t frame) { @@ -47,8 +48,10 @@ void define_ClusterFinderMT(py::module &m, const std::string &typestr) { [](ClusterFinderMT &self, py::array_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( diff --git a/src/ClusterFinderMT.test.cpp b/src/ClusterFinderMT.test.cpp index e890481..4bf7aec 100644 --- a/src/ClusterFinderMT.test.cpp +++ b/src/ClusterFinderMT.test.cpp @@ -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( - 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; + + 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 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 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); +}