From 30ba8a7983aae001153d559c757b287255f17488 Mon Sep 17 00:00:00 2001 From: VzdornovNA88 Date: Fri, 2 Jan 2026 14:07:40 +0500 Subject: [PATCH] Replace SPSC lock free queue (LockFreeQueue) with lock based queue (SafeQueue) because MPSC cases are in the isobus library --- .../can_hardware_interface.hpp | 4 +- test/CMakeLists.txt | 1 + test/utility_queue_tests.cpp | 156 ++++++++++++++++++ .../isobus/utility/thread_synchronization.hpp | 154 ++++++++++++++++- 4 files changed, 307 insertions(+), 8 deletions(-) create mode 100644 test/utility_queue_tests.cpp diff --git a/hardware_integration/include/isobus/hardware_integration/can_hardware_interface.hpp b/hardware_integration/include/isobus/hardware_integration/can_hardware_interface.hpp index 0ba1f7e70..21adabb5a 100644 --- a/hardware_integration/include/isobus/hardware_integration/can_hardware_interface.hpp +++ b/hardware_integration/include/isobus/hardware_integration/can_hardware_interface.hpp @@ -154,8 +154,8 @@ namespace isobus std::shared_ptr frameHandler; ///< The CAN driver to use for a CAN channel - LockFreeQueue messagesToBeTransmittedQueue; ///< Transmit message queue for a CAN channel - LockFreeQueue receivedMessagesQueue; ///< Receive message queue for a CAN channel + Queue messagesToBeTransmittedQueue; ///< Transmit message queue for a CAN channel + Queue receivedMessagesQueue; ///< Receive message queue for a CAN channel }; /// @brief Singleton instance of the CANHardwareInterface class diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index a69f424b1..db693d00e 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -36,6 +36,7 @@ set(TEST_INCLUDE helpers/control_function_helpers.hpp # Set test source files set(TEST_SRC + utility_queue_tests.cpp core_network_management_tests.cpp identifier_tests.cpp transport_protocol_tests.cpp diff --git a/test/utility_queue_tests.cpp b/test/utility_queue_tests.cpp new file mode 100644 index 000000000..03e2b3a70 --- /dev/null +++ b/test/utility_queue_tests.cpp @@ -0,0 +1,156 @@ +#include +#include +#include +#include +#include +#include +#include +#include +class QUEUE_TESTS : public ::testing::Test +{ +protected: + static constexpr std::size_t QUEUE_SIZE = 500; + Queue queue{ QUEUE_SIZE }; +}; + +TEST_F(QUEUE_TESTS, MultipleProducersMultipleConsumersStressTest) +{ + const int NUM_PRODUCERS = 16; + const int NUM_CONSUMERS = 4; + const int ITEMS_PER_PRODUCER = 100000; // Reduced for faster testing + const int TOTAL_ITEMS = NUM_PRODUCERS * ITEMS_PER_PRODUCER; + + std::atomic produced_count{ 0 }; + std::atomic consumed_count{ 0 }; + std::vector> producer_done(NUM_PRODUCERS); + for (auto &done : producer_done) + done = false; + std::vector producers; + + // Producers + for (int p = 0; p < NUM_PRODUCERS; ++p) + { + producers.emplace_back([this, ITEMS_PER_PRODUCER, &produced_count, &producer_done, p]() { + for (int i = 0; i < ITEMS_PER_PRODUCER; ++i) + { + while (!queue.push(i)) + { + // Spin if queue is full + std::this_thread::yield(); + } + produced_count++; + } + producer_done[p] = true; // Mark this producer as done + }); + } + + // Multiple Consumers + std::vector consumers; + for (int c = 0; c < NUM_CONSUMERS; ++c) + { + consumers.emplace_back([this, &consumed_count, &producer_done]() { + while (true) + { + bool all_producers_done = true; + for (const auto &done : producer_done) + { + if (!done.load()) + { + all_producers_done = false; + break; + } + } + + if (queue.pop()) + { + consumed_count++; + } + else if (all_producers_done) + { + break; // All producers done and queue is empty + } + else + { + std::this_thread::yield(); + } + } + }); + } + + // Wait for producers + for (auto &t : producers) + { + t.join(); + } + + // Wait for consumers + for (auto &t : consumers) + { + t.join(); + } + + // Check if all items were produced + EXPECT_EQ(produced_count.load(), TOTAL_ITEMS); + // Due to race conditions, consumed_count might be less than produced if data is overwritten + std::cout << "Produced: " << produced_count.load() << ", Consumed: " << consumed_count.load() << std::endl; + + // This assertion may fail due to race conditions + EXPECT_EQ(consumed_count.load(), TOTAL_ITEMS); +} + +// Test all methods of the queue +TEST_F(QUEUE_TESTS, QueueAPIMethodsTest) +{ + // Test 1: Basic push/pop operations + EXPECT_TRUE(queue.push(1)); + EXPECT_TRUE(queue.push(2)); + EXPECT_TRUE(queue.push(3)); + + EXPECT_EQ(queue.size(), 3); + EXPECT_FALSE(queue.is_empty()); + + // Test 2: peek method + int peek_value = 0; + EXPECT_TRUE(queue.peek(peek_value)); + EXPECT_EQ(peek_value, 1); // Should be first item + + // Test 3: pop() without parameter + EXPECT_TRUE(queue.pop()); + EXPECT_EQ(queue.size(), 2); + + // Test 4: pop(value_type*) method + int popped_value1 = 0; + EXPECT_TRUE(queue.pop(&popped_value1)); + EXPECT_EQ(popped_value1, 2); + EXPECT_EQ(queue.size(), 1); + + // Test 5: pop(value_type&) method + int popped_value2 = 0; + EXPECT_TRUE(queue.pop(popped_value2)); + EXPECT_EQ(popped_value2, 3); + EXPECT_EQ(queue.size(), 0); + + // Test 6: Empty queue checks + EXPECT_TRUE(queue.is_empty()); + int temp = 0; + EXPECT_FALSE(queue.peek(temp)); + EXPECT_FALSE(queue.pop()); + EXPECT_FALSE(queue.pop(&temp)); + EXPECT_FALSE(queue.pop(temp)); + + // Test 7: Clear method + EXPECT_TRUE(queue.push(10)); + EXPECT_TRUE(queue.push(20)); + EXPECT_EQ(queue.size(), 2); + + queue.clear(); + EXPECT_EQ(queue.size(), 0); + EXPECT_TRUE(queue.is_empty()); + + // Test 8: Move semantics + int moved_value = 42; + EXPECT_TRUE(queue.push(std::move(moved_value))); + int result = 0; + EXPECT_TRUE(queue.pop(result)); + EXPECT_EQ(result, 42); +} diff --git a/utility/include/isobus/utility/thread_synchronization.hpp b/utility/include/isobus/utility/thread_synchronization.hpp index fbdec7cbc..1826b6ba3 100644 --- a/utility/include/isobus/utility/thread_synchronization.hpp +++ b/utility/include/isobus/utility/thread_synchronization.hpp @@ -107,11 +107,11 @@ namespace isobus /// @brief A template class for a lock free queue. /// @tparam T The item type for the queue. template -class LockFreeQueue +class SPSCLockFreeQueue { public: /// @brief Constructor for the lock free queue. - explicit LockFreeQueue(std::size_t size) : + explicit SPSCLockFreeQueue(std::size_t size) : buffer(size), capacity(size) { // Validate the size of the queue, if assertion is disabled, set the size to 1. @@ -210,12 +210,52 @@ class UnsafeQueue public: using value_type = T; + /// @brief Constructor for the queue. + /// @param size For backward compatibility. + explicit UnsafeQueue(std::size_t size) + { + (void)size; + } + + UnsafeQueue() = default; + template::value>::type> - void push(U &&item) + bool push(U &&item) { queue.push(std::forward(item)); + return true; } + /// @brief Peek at the next item in the queue. + /// @param item The item to peek at in the queue. + /// @return True if the item was peeked at in the queue, false if the queue is empty. + bool peek(value_type &item) + { + if (queue.empty()) + { + return false; + } + + item = queue.front(); + return true; + } + + /// @brief Pop an item from the queue. + /// @return True if the item was popped from the queue, false if the queue is empty. + bool pop() + { + if (queue.empty()) + { + return false; + } + + queue.pop(); + return true; + } + + /// @brief Pop an item from the queue and return it. + /// @param item Pointer to store the popped item. + /// @return True if the item was popped from the queue, false if the queue is empty. bool pop(value_type *item) { if (queue.empty()) @@ -227,6 +267,42 @@ class UnsafeQueue return true; } + /// @brief Pop an item from the queue and return it. + /// @param item Reference to store the popped item. + /// @return True if the item was popped from the queue, false if the queue is empty. + bool pop(value_type &item) + { + if (queue.empty()) + { + return false; + } + item = std::move(queue.front()); + queue.pop(); + return true; + } + + /// @brief Check if the queue is full. + /// @return Always returns false, since this version of the queue is not limited in size. + bool is_full() const + { + return false; + } + + /// @brief Check if the queue is empty. + /// @return True if the queue is empty, false otherwise. + bool is_empty() const + { + return queue.empty(); + } + + /// @brief Get the number of items in the queue. + /// @return The number of items in the queue. + std::size_t size() const + { + return queue.size(); + } + + /// @brief Clear the queue. void clear() { queue = {}; @@ -251,19 +327,84 @@ class SafeQueue : private UnsafeQueue public: using value_type = T; + /// @brief Constructor for the safe queue. + /// @param size For backward compatibility. + explicit SafeQueue(std::size_t size) : + Q(size) {} + + SafeQueue() = default; + + /// @brief Push an item to the queue + /// @tparam U The type of the item to push (must be convertible to value_type). + /// @param item The item to push to the queue. + /// @return True if the item was pushed to the queue. template::value>::type> - void push(U &&item) + bool push(U &&item) + { + std::lock_guard lock(mtx); + return Q::push(std::forward(item)); + } + + /// @brief Peek at the next item in the queue. + /// @param item The item to peek at in the queue. + /// @return True if the item was peeked at in the queue, false if the queue is empty. + bool peek(value_type &item) + { + std::lock_guard lock(mtx); + return Q::peek(item); + } + + /// @brief Pop an item from the queue. + /// @return True if the item was popped from the queue, false if the queue is empty. + bool pop() { std::lock_guard lock(mtx); - Q::push(std::forward(item)); + return Q::pop(); } + /// @brief Pop an item from the queue and return it. + /// @param item Pointer to store the popped item. + /// @return True if the item was popped from the queue, false if the queue is empty. bool pop(value_type *item) { std::lock_guard lock(mtx); return Q::pop(item); } + /// @brief Pop an item from the queue and return it. + /// @param item Reference to store the popped item. + /// @return True if the item was popped from the queue, false if the queue is empty. + bool pop(value_type &item) + { + std::lock_guard lock(mtx); + return Q::pop(item); + } + + /// @brief Check if the queue is full. + /// @return Always returns false, since this version of the queue is not limited in size.(For backward compatibility.) + bool is_full() const + { + std::lock_guard lock(mtx); + return Q::is_full(); + } + + /// @brief Check if the queue is empty. + /// @return True if the queue is empty, false otherwise. + bool is_empty() const + { + std::lock_guard lock(mtx); + return Q::is_empty(); + } + + /// @brief Get the number of items in the queue. + /// @return The number of items in the queue. + std::size_t size() const + { + std::lock_guard lock(mtx); + return Q::size(); + } + + /// @brief Clear the queue. void clear() { std::lock_guard lock(mtx); @@ -271,8 +412,9 @@ class SafeQueue : private UnsafeQueue } private: - std::mutex mtx; + mutable std::mutex mtx; }; + template using Queue = SafeQueue; #endif