Merge branch 'developer' into dev/servers-opt

slsReceiverSoftware/include/sls/CircularFifo.h

@@ -9,8 +9,10 @@
  * modified by the sls detector group
  * */
 
-#include <iostream>
-#include <semaphore.h>
+#include "sls/counting_semaphore.h"
+
+#include <atomic>
+#include <cstddef>
 #include <vector>
 
 namespace sls {
@@ -23,23 +25,21 @@ template <typename Element> class CircularFifo {
     size_t head{0};
     size_t capacity;
     std::vector<Element *> 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<int> 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<std::ptrdiff_t>(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 <typename Element> class CircularFifo {
 };
 
 template <typename Element> int CircularFifo<Element>::getDataValue() const {
-    int value;
-    sem_getvalue(&data_mutex, &value);
-    return value;
+    return count_.load(std::memory_order_relaxed);
 }
 
 template <typename Element> int CircularFifo<Element>::getFreeValue() const {
-    int value;
-    sem_getvalue(&free_mutex, &value);
-    return value;
+    return static_cast<int>(capacity - 1) - getDataValue();
 }
 
 /** Producer only: Adds item to the circular queue.
@@ -72,14 +68,16 @@ template <typename Element> int CircularFifo<Element>::getFreeValue() const {
  * \return whether operation was successful or not */
 template <typename Element>
 bool CircularFifo<Element>::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<Element>::push(Element *&item, bool no_block) {
  * \return whether operation was successful or not */
 template <typename Element>
 bool CircularFifo<Element>::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<Element>::pop(Element *&item, bool no_block) {
  *
  * \return true if circular buffer is empty */
 template <typename Element> bool CircularFifo<Element>::isEmpty() const {
-    return (getDataValue() == 0);
+    return getDataValue() == 0;
 }
 
 /** Useful for testing and Producer check of status
@@ -118,18 +118,7 @@ template <typename Element> bool CircularFifo<Element>::isEmpty() const {
  *
  * \return true if circular buffer is full.  */
 template <typename Element> bool CircularFifo<Element>::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 <typename Element>
-size_t CircularFifo<Element>::increment(size_t i) const {
-    i = (i + 1) % capacity;
-    return i;
-}
-
-} // namespace sls
+} // namespace sls

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<std::mutex> 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<std::mutex> 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<sem_t> semaphores;
@@ -579,10 +578,9 @@ int main(int argc, char *argv[]) {
     {
         std::lock_guard<std::mutex> lock(stat.mtx);
         stat.terminate = true;
-        sem_post(&stat.available);
+        stat.available.release();
     }
     combinerThread.join();
-    sem_destroy(&stat.available);
 
     if (threadException) {
         try {

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;

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 <atomic>
-#include <semaphore.h>
 #include <string>
+#include <sys/types.h> // pid_t
 #include <thread>
 
 namespace sls {
@@ -45,7 +46,7 @@ class ThreadObject : private virtual slsDetectorDefs {
     std::atomic<bool> killThread{false};
     std::atomic<bool> runningFlag{false};
     std::thread threadObject;
-    sem_t semaphore;
+    counting_semaphore<> semaphore{0};
     const std::string type;
     std::atomic<pid_t> threadId{0};
 };

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 <semaphore> types so that the project can switch by
+ * replacing this include with <semaphore> 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 <condition_variable>
+#include <cstddef>
+#include <limits>
+#include <mutex>
+
+namespace sls {
+
+template <std::ptrdiff_t LeastMaxValue =
+              std::numeric_limits<std::ptrdiff_t>::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<std::mutex> lk(mtx_);
+        cv_.wait(lk, [this] { return count_ > 0; });
+        --count_;
+    }
+
+    bool try_acquire() noexcept {
+        std::lock_guard<std::mutex> 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<std::mutex> 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

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

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 <atomic>
+#include <chrono>
+#include <cstddef>
+#include <thread>
+#include <type_traits>
+#include <vector>
+
+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<counting_semaphore<>>::value,
+              "semaphore must not be copyable");
+static_assert(!std::is_copy_assignable<counting_semaphore<>>::value,
+              "semaphore must not be copyable");
+static_assert(!std::is_move_constructible<counting_semaphore<>>::value,
+              "semaphore must not be movable");
+static_assert(
+    std::is_same<binary_semaphore, counting_semaphore<1>>::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<bool> 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<int> acquired{0};
+    std::vector<std::thread> 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<int> acquired{0};
+    std::vector<std::thread> 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