From 5ee7844b350c4453bdb3fd617d625229b1614c79 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Erik=20Fr=C3=B6jdh?= Date: Mon, 26 Oct 2026 18:03:13 +0100 Subject: [PATCH] dev/ Replaced sem_t with a C++20 like counting_semaphore (#1587) * added C++17 backport of std::counting_semaphore sls::counting_semaphore and sls::binary_semaphore mirror the C++20 types and are built on std::mutex and std::condition_variable. Unnamed POSIX semaphores (sem_init) are not implemented on macOS. --- .../include/sls/CircularFifo.h | 79 ++++------ .../src/FrameSynchronizerApp.cpp | 18 +-- slsReceiverSoftware/src/ThreadObject.cpp | 8 +- slsReceiverSoftware/src/ThreadObject.h | 5 +- .../include/sls/counting_semaphore.h | 80 ++++++++++ slsSupportLib/tests/CMakeLists.txt | 1 + .../tests/test-counting_semaphore.cpp | 147 ++++++++++++++++++ 7 files changed, 276 insertions(+), 62 deletions(-) create mode 100644 slsSupportLib/include/sls/counting_semaphore.h create mode 100644 slsSupportLib/tests/test-counting_semaphore.cpp diff --git a/slsReceiverSoftware/include/sls/CircularFifo.h b/slsReceiverSoftware/include/sls/CircularFifo.h index 03f867882..ca9dec57b 100644 --- a/slsReceiverSoftware/include/sls/CircularFifo.h +++ b/slsReceiverSoftware/include/sls/CircularFifo.h @@ -9,8 +9,10 @@ * modified by the sls detector group * */ -#include -#include +#include "sls/counting_semaphore.h" + +#include +#include #include namespace sls { @@ -23,23 +25,21 @@ template class CircularFifo { size_t head{0}; size_t capacity; std::vector data; - mutable sem_t data_mutex; - mutable sem_t free_mutex; - size_t increment(size_t i) const; + counting_semaphore<> data_sem; // # of items available to read + counting_semaphore<> free_sem; // # of slots available to write + std::atomic count_{0}; // current number of items, for diagnostics + + size_t increment(size_t i) const { return (i + 1) % capacity; } public: - explicit CircularFifo(size_t size) : capacity(size + 1), data(capacity) { - sem_init(&data_mutex, 0, 0); - sem_init(&free_mutex, 0, size); - } + explicit CircularFifo(size_t size) + : capacity(size + 1), data(capacity), data_sem(0), + free_sem(static_cast(size)) {} CircularFifo(const CircularFifo &) = delete; CircularFifo(CircularFifo &&) = delete; - virtual ~CircularFifo() { - sem_destroy(&data_mutex); - sem_destroy(&free_mutex); - } + virtual ~CircularFifo() = default; bool push(Element *&item, bool no_block = false); bool pop(Element *&item, bool no_block = false); @@ -52,15 +52,11 @@ template class CircularFifo { }; template int CircularFifo::getDataValue() const { - int value; - sem_getvalue(&data_mutex, &value); - return value; + return count_.load(std::memory_order_relaxed); } template int CircularFifo::getFreeValue() const { - int value; - sem_getvalue(&free_mutex, &value); - return value; + return static_cast(capacity - 1) - getDataValue(); } /** Producer only: Adds item to the circular queue. @@ -72,14 +68,16 @@ template int CircularFifo::getFreeValue() const { * \return whether operation was successful or not */ template bool CircularFifo::push(Element *&item, bool no_block) { - // check for fifo full - if (no_block && isFull()) - return false; - - sem_wait(&free_mutex); + if (no_block) { + if (!free_sem.try_acquire()) + return false; + } else { + free_sem.acquire(); + } data[tail] = item; tail = increment(tail); - sem_post(&data_mutex); + count_.fetch_add(1, std::memory_order_relaxed); + data_sem.release(); return true; } @@ -92,14 +90,16 @@ bool CircularFifo::push(Element *&item, bool no_block) { * \return whether operation was successful or not */ template bool CircularFifo::pop(Element *&item, bool no_block) { - // check for fifo empty - if (no_block && isEmpty()) - return false; - - sem_wait(&data_mutex); + if (no_block) { + if (!data_sem.try_acquire()) + return false; + } else { + data_sem.acquire(); + } item = data[head]; head = increment(head); - sem_post(&free_mutex); + count_.fetch_sub(1, std::memory_order_relaxed); + free_sem.release(); return true; } @@ -109,7 +109,7 @@ bool CircularFifo::pop(Element *&item, bool no_block) { * * \return true if circular buffer is empty */ template bool CircularFifo::isEmpty() const { - return (getDataValue() == 0); + return getDataValue() == 0; } /** Useful for testing and Producer check of status @@ -118,18 +118,7 @@ template bool CircularFifo::isEmpty() const { * * \return true if circular buffer is full. */ template bool CircularFifo::isFull() const { - return (getFreeValue() == 0); + return getFreeValue() == 0; } -/** Increment helper function for index of the circular queue - * index is incremented or wrapped - * - * \param i the index to the incremented/wrapped - * \return new value for the index */ -template -size_t CircularFifo::increment(size_t i) const { - i = (i + 1) % capacity; - return i; -} - -} // namespace sls \ No newline at end of file +} // namespace sls diff --git a/slsReceiverSoftware/src/FrameSynchronizerApp.cpp b/slsReceiverSoftware/src/FrameSynchronizerApp.cpp index d3b8c932f..014ab0d10 100644 --- a/slsReceiverSoftware/src/FrameSynchronizerApp.cpp +++ b/slsReceiverSoftware/src/FrameSynchronizerApp.cpp @@ -9,6 +9,7 @@ #include "sls/Receiver.h" #include "sls/ToString.h" #include "sls/container_utils.h" +#include "sls/counting_semaphore.h" #include "sls/logger.h" #include "sls/network_utils.h" #include "sls/sls_detector_defs.h" @@ -45,16 +46,14 @@ struct FrameStatus { bool starting = true; bool terminate = false; int num_receivers; - sem_t available; + sls::counting_semaphore<> available{0}; std::mutex mtx; ZmqMsgList headers; PortFrameMap frames; ZmqMsgList ends; FrameStatus(bool start, bool term, int num_recv) - : starting(start), terminate(term), num_receivers(num_recv) { - sem_init(&available, 0, 0); - } + : starting(start), terminate(term), num_receivers(num_recv) {} }; FrameStatus *global_frame_status = nullptr; @@ -165,7 +164,7 @@ void Correlate(FrameStatus *stat) { } while (true) { - sem_wait(&(stat->available)); + stat->available.acquire(); { std::lock_guard lock(stat->mtx); @@ -328,7 +327,7 @@ void StartAcquisitionCallback( stat->frames[port].clear(); } } - sem_post(&stat->available); + stat->available.release(); } void AcquisitionFinishedCallback( @@ -366,7 +365,7 @@ void AcquisitionFinishedCallback( std::lock_guard lock(stat->mtx); stat->ends.push_back(hmsg); } - sem_post(&stat->available); + stat->available.release(); } void GetDataCallback(slsDetectorDefs::sls_receiver_header &header, @@ -487,7 +486,7 @@ void GetDataCallback(slsDetectorDefs::sls_receiver_header &header, stat->frames[callbackHeader.udpPort][header.detHeader.frameNumber] .push_back(msg); } - sem_post(&stat->available); + stat->available.release(); } std::vector semaphores; @@ -579,10 +578,9 @@ int main(int argc, char *argv[]) { { std::lock_guard lock(stat.mtx); stat.terminate = true; - sem_post(&stat.available); + stat.available.release(); } combinerThread.join(); - sem_destroy(&stat.available); if (threadException) { try { diff --git a/slsReceiverSoftware/src/ThreadObject.cpp b/slsReceiverSoftware/src/ThreadObject.cpp index 746a03c28..3ce6a5718 100644 --- a/slsReceiverSoftware/src/ThreadObject.cpp +++ b/slsReceiverSoftware/src/ThreadObject.cpp @@ -16,7 +16,6 @@ namespace sls { ThreadObject::ThreadObject(int index, std::string type) : index(index), type(type) { LOG(logDEBUG) << type << " thread created: " << index; - sem_init(&semaphore, 1, 0); try { threadObject = std::thread(&ThreadObject::RunningThread, this); } catch (...) { @@ -27,9 +26,8 @@ ThreadObject::ThreadObject(int index, std::string type) ThreadObject::~ThreadObject() { killThread = true; - sem_post(&semaphore); + semaphore.release(); threadObject.join(); - sem_destroy(&semaphore); } pid_t ThreadObject::GetThreadId() const { return threadId; } @@ -49,14 +47,14 @@ void ThreadObject::RunningThread() { ThreadExecution(); } // wait till the next acquisition - sem_wait(&semaphore); + semaphore.acquire(); } LOG(logINFOBLUE) << "Exiting [ " << type << " Thread " << index << ", Tid: " << threadId << "]"; threadId = 0; } -void ThreadObject::Continue() { sem_post(&semaphore); } +void ThreadObject::Continue() { semaphore.release(); } void ThreadObject::SetThreadPriority(int priority) { struct sched_param param; diff --git a/slsReceiverSoftware/src/ThreadObject.h b/slsReceiverSoftware/src/ThreadObject.h index 6089e1042..3f1418834 100644 --- a/slsReceiverSoftware/src/ThreadObject.h +++ b/slsReceiverSoftware/src/ThreadObject.h @@ -9,12 +9,13 @@ *@short creates/destroys a thread */ +#include "sls/counting_semaphore.h" #include "sls/logger.h" #include "sls/sls_detector_defs.h" #include -#include #include +#include // pid_t #include namespace sls { @@ -45,7 +46,7 @@ class ThreadObject : private virtual slsDetectorDefs { std::atomic killThread{false}; std::atomic runningFlag{false}; std::thread threadObject; - sem_t semaphore; + counting_semaphore<> semaphore{0}; const std::string type; std::atomic threadId{0}; }; diff --git a/slsSupportLib/include/sls/counting_semaphore.h b/slsSupportLib/include/sls/counting_semaphore.h new file mode 100644 index 000000000..2f631e06d --- /dev/null +++ b/slsSupportLib/include/sls/counting_semaphore.h @@ -0,0 +1,80 @@ +// SPDX-License-Identifier: LGPL-3.0-or-other +// Copyright (C) 2021 Contributors to the SLS Detector Package +#pragma once + +/** + * @file counting_semaphore.h + * @short C++17 backport of std::counting_semaphore and std::binary_semaphore + * + * Mirrors the C++20 types so that the project can switch by + * replacing this include with and sls:: with std::. Unnamed POSIX + * semaphores (sem_init) cannot be used instead as they are not implemented on + * macOS. + * + * Differences to the std types: + * - try_acquire_for and try_acquire_until are not implemented + * - the constructor is not constexpr + * - try_acquire never fails spuriously (std allows it to) + * - the preconditions (0 <= desired <= max() and + * 0 <= update <= max() - counter) are not checked, as in std violating + * them is undefined behaviour + * + * Built on std::mutex and std::condition_variable, therefore NOT + * async-signal-safe. Do not call release() from a signal handler. + */ + +#include +#include +#include +#include + +namespace sls { + +template ::max()> +class counting_semaphore { + static_assert(LeastMaxValue >= 0, "LeastMaxValue must be non-negative"); + + public: + static constexpr std::ptrdiff_t max() noexcept { return LeastMaxValue; } + + explicit counting_semaphore(std::ptrdiff_t desired) : count_(desired) {} + + counting_semaphore(const counting_semaphore &) = delete; + counting_semaphore &operator=(const counting_semaphore &) = delete; + + void acquire() { + std::unique_lock lk(mtx_); + cv_.wait(lk, [this] { return count_ > 0; }); + --count_; + } + + bool try_acquire() noexcept { + std::lock_guard lk(mtx_); + if (count_ == 0) + return false; + --count_; + return true; + } + + void release(std::ptrdiff_t update = 1) { + // notify while holding the lock: a thread returning from acquire() + // may destroy the semaphore, which must not happen before we are done + // with the condition variable + std::lock_guard lk(mtx_); + count_ += update; + if (update == 1) + cv_.notify_one(); + else + cv_.notify_all(); + } + + private: + std::mutex mtx_; + std::condition_variable cv_; + std::ptrdiff_t count_; +}; + +using binary_semaphore = counting_semaphore<1>; + +} // namespace sls diff --git a/slsSupportLib/tests/CMakeLists.txt b/slsSupportLib/tests/CMakeLists.txt index 827db97ff..d5423d36f 100755 --- a/slsSupportLib/tests/CMakeLists.txt +++ b/slsSupportLib/tests/CMakeLists.txt @@ -4,6 +4,7 @@ target_sources(tests PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/test-bit_utils.cpp ${CMAKE_CURRENT_SOURCE_DIR}/test-file_utils.cpp ${CMAKE_CURRENT_SOURCE_DIR}/test-container_utils.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/test-counting_semaphore.cpp ${CMAKE_CURRENT_SOURCE_DIR}/test-network_utils.cpp ${CMAKE_CURRENT_SOURCE_DIR}/test-string_utils.cpp ${CMAKE_CURRENT_SOURCE_DIR}/test-Timer.cpp diff --git a/slsSupportLib/tests/test-counting_semaphore.cpp b/slsSupportLib/tests/test-counting_semaphore.cpp new file mode 100644 index 000000000..6b36a17a6 --- /dev/null +++ b/slsSupportLib/tests/test-counting_semaphore.cpp @@ -0,0 +1,147 @@ +// SPDX-License-Identifier: LGPL-3.0-or-other +// Copyright (C) 2021 Contributors to the SLS Detector Package +#include "catch.hpp" +#include "sls/counting_semaphore.h" + +#include +#include +#include +#include +#include +#include + +namespace sls { + +// Only what std::counting_semaphore guarantees is tested, so that the tests +// also pass with the std types once the project moves to C++20. + +static_assert(!std::is_copy_constructible>::value, + "semaphore must not be copyable"); +static_assert(!std::is_copy_assignable>::value, + "semaphore must not be copyable"); +static_assert(!std::is_move_constructible>::value, + "semaphore must not be movable"); +static_assert( + std::is_same>::value, + "binary_semaphore is a counting_semaphore with a least max value of 1"); + +TEST_CASE("max is at least the least max value", "[support][semaphore]") { + CHECK(binary_semaphore::max() >= 1); + CHECK(counting_semaphore<5>::max() >= 5); + CHECK(counting_semaphore<>::max() >= 1); +} + +TEST_CASE("try_acquire takes what the semaphore was constructed with", + "[support][semaphore]") { + counting_semaphore<> s(2); + CHECK(s.try_acquire()); + CHECK(s.try_acquire()); + CHECK_FALSE(s.try_acquire()); +} + +TEST_CASE("try_acquire fails on an empty semaphore", "[support][semaphore]") { + binary_semaphore s{0}; + CHECK_FALSE(s.try_acquire()); + s.release(); + CHECK(s.try_acquire()); + CHECK_FALSE(s.try_acquire()); +} + +TEST_CASE("release with an update count adds that many", + "[support][semaphore]") { + counting_semaphore<> s(0); + s.release(3); + s.release(); + for (int i = 0; i != 4; ++i) { + CHECK(s.try_acquire()); + } + CHECK_FALSE(s.try_acquire()); +} + +TEST_CASE("acquire does not block when the count is positive", + "[support][semaphore]") { + counting_semaphore<> s(1); + s.acquire(); + CHECK_FALSE(s.try_acquire()); +} + +TEST_CASE("acquire blocks until release", "[support][semaphore]") { + binary_semaphore s{0}; + std::atomic acquired{false}; + std::thread t([&] { + s.acquire(); + acquired = true; + }); + + // cannot have passed acquire, no matter how the thread is scheduled + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + CHECK_FALSE(acquired); + + s.release(); + t.join(); + CHECK(acquired); + CHECK_FALSE(s.try_acquire()); +} + +TEST_CASE("release with an update count wakes up that many waiting threads", + "[support][semaphore]") { + const int nThreads = 4; + counting_semaphore<> s(0); + std::atomic acquired{0}; + std::vector threads; + for (int i = 0; i != nThreads; ++i) { + threads.emplace_back([&] { + s.acquire(); + ++acquired; + }); + } + s.release(nThreads); + for (auto &t : threads) { + t.join(); + } + CHECK(acquired == nThreads); + CHECK_FALSE(s.try_acquire()); +} + +TEST_CASE("every release is acquired exactly once", "[support][semaphore]") { + const int nProducers = 3; + const int nConsumers = 3; + const int nPerThread = 2000; + counting_semaphore<> s(0); + std::atomic acquired{0}; + std::vector threads; + for (int i = 0; i != nConsumers; ++i) { + threads.emplace_back([&] { + for (int j = 0; j != nPerThread; ++j) { + s.acquire(); + ++acquired; + } + }); + } + for (int i = 0; i != nProducers; ++i) { + threads.emplace_back([&] { + for (int j = 0; j != nPerThread; ++j) { + s.release(); + } + }); + } + for (auto &t : threads) { + t.join(); + } + CHECK(acquired == nConsumers * nPerThread); + CHECK_FALSE(s.try_acquire()); +} + +TEST_CASE("a waiter can destroy the semaphore as soon as it has acquired", + "[support][semaphore]") { + for (int i = 0; i != 200; ++i) { + auto *done = new binary_semaphore(0); + std::thread worker([done] { done->release(); }); + done->acquire(); + delete done; + worker.join(); + } + SUCCEED(); +} + +} // namespace sls