diff --git a/score/launch_manager/src/daemon/src/common/concurrency/BUILD b/score/launch_manager/src/daemon/src/common/concurrency/BUILD index 08ab3ba30..efecb3822 100644 --- a/score/launch_manager/src/daemon/src/common/concurrency/BUILD +++ b/score/launch_manager/src/daemon/src/common/concurrency/BUILD @@ -25,6 +25,29 @@ cc_library( ], ) +cc_library( + name = "fixed_size_queue", + hdrs = [ + "fixed_size_queue.hpp", + ], + include_prefix = "score/mw/launch_manager/common/concurrency", + strip_include_prefix = "/score/launch_manager/src/daemon/src/common/concurrency", + visibility = ["//score:__subpackages__"], + deps = [ + "@score_baselibs//score/language/futurecpp", + ], +) + +cc_test( + name = "fixed_size_queue_test", + srcs = ["fixed_size_queue_test.cpp"], + visibility = ["//tests:__subpackages__"], + deps = [ + ":fixed_size_queue", + "@googletest//:gtest_main", + ], +) + cc_library( name = "mpmc_concurrent_queue", hdrs = [ @@ -68,6 +91,7 @@ cc_library( strip_include_prefix = "/score/launch_manager/src/daemon/src/common/concurrency", visibility = ["//score:__subpackages__"], deps = [ + ":fixed_size_queue", "@score_baselibs//score/language/futurecpp", ], ) diff --git a/score/launch_manager/src/daemon/src/common/concurrency/fixed_size_queue.hpp b/score/launch_manager/src/daemon/src/common/concurrency/fixed_size_queue.hpp new file mode 100644 index 000000000..d884a5bcf --- /dev/null +++ b/score/launch_manager/src/daemon/src/common/concurrency/fixed_size_queue.hpp @@ -0,0 +1,115 @@ +/******************************************************************************** + * Copyright (c) 2026 Contributors to the Eclipse Foundation + * + * See the NOTICE file(s) distributed with this work for additional + * information regarding copyright ownership. + * + * This program and the accompanying materials are made available under the + * terms of the Apache License Version 2.0 which is available at + * https://www.apache.org/licenses/LICENSE-2.0 + * + * SPDX-License-Identifier: Apache-2.0 + ********************************************************************************/ + +#ifndef FIXED_SIZED_QUEUE_HPP_INCLUDE +#define FIXED_SIZED_QUEUE_HPP_INCLUDE + +#include +#include +#include + +namespace score::lcm::internal +{ + +/// @brief Fixed-size FIFO queue +/// @details Uses std::optional to eliminate default construction of +/// type T elements at construction +/// @tparam T The type of elements stored in the queue. +/// Supports move-only and copy-only types. +template +class FixedSizeQueue +{ + public: + /// @brief Constructs a FixedSizeQueue with zero capacity. + /// @details push will always return false + /// tryPop will always return std::nullopt + FixedSizeQueue() : capacity_{0U}, slots_{} {} + + /// @brief Constructs a FixedSizeQueue with a fixed runtime capacity. + /// @details If the specified size is 0, it is equivalent + /// to a default constructed FixedSizeQueue. + /// @param size The desired maximum number of elements. + explicit FixedSizeQueue(std::size_t size) : capacity_(size) { + slots_.resize(capacity_); + } + + /// @brief Inserts a new element directly at the tail of the queue. + /// @tparam Args Variadic template arguments forwarded to the constructor of T. + /// @param args The arguments used to construct the object of type T. + /// @return true if the element was successfully inserted; false if the queue is full (overflow protection). + template + bool push(Args&&... args) { + if(full()) { + return false; + } + + slots_[tail_].emplace(std::forward(args)...); + tail_ = (tail_ + 1U) % capacity_; + count_++; + + return true; + } + + + /// @brief Attempts to extract and remove the oldest element from the queue. + /// @details The popped element is immediately destroyed in the internal + /// buffer to free resources. + /// @return A std::optional containing the element, + /// or std::nullopt if the queue was empty (underflow protection). + std::optional tryPop() { + if (empty()) + { + return std::nullopt; + } + + std::optional item = std::move(slots_[head_]); + slots_[head_] = std::nullopt; + head_ = (head_ + 1U) % capacity_; + count_--; + + return item; + } + + /// @brief Checks if the queue contains no elements. + /// @return true if empty, otherwise false. + bool empty() const { return (count_ == 0U); } + + /// @brief Checks if the queue has reached its maximum capacity. + /// @return true if full, otherwise false. + bool full() const { return (count_ >= capacity_); } + + /// @brief Retrieves the current number of active elements in the queue. + /// @return The count of stored elements. + std::size_t size() const { return count_;}; + + /// @brief Retrieves the capacity of the queue. + /// @return The capacity of the queue + std::size_t capacity() const { return capacity_;}; + + + private: + /// @brief Index of the oldest element in the buffer (read index). + std::size_t head_ = 0U; + /// @brief Index where the next element will be inserted (write index). + std::size_t tail_ = 0U; + /// @brief Current number of active elements in the queue. + std::size_t count_ = 0U; + /// @brief Maximum capacity of the queue. + std::size_t capacity_; + /// @brief Internal buffer holding the optional elements. + std::vector> slots_; +}; + +} // namespace score::lcm::internal + +#endif // FIXED_SIZED_QUEUE_HPP_INCLUDE diff --git a/score/launch_manager/src/daemon/src/common/concurrency/fixed_size_queue_test.cpp b/score/launch_manager/src/daemon/src/common/concurrency/fixed_size_queue_test.cpp new file mode 100644 index 000000000..7b5b8d0ed --- /dev/null +++ b/score/launch_manager/src/daemon/src/common/concurrency/fixed_size_queue_test.cpp @@ -0,0 +1,208 @@ +/******************************************************************************** + * Copyright (c) 2026 Contributors to the Eclipse Foundation + * + * See the NOTICE file(s) distributed with this work for additional + * information regarding copyright ownership. + * + * This program and the accompanying materials are made available under the + * terms of the Apache License Version 2.0 which is available at + * https://www.apache.org/licenses/LICENSE-2.0 + * + * SPDX-License-Identifier: Apache-2.0 + ********************************************************************************/ + +#include "score/mw/launch_manager/common/concurrency/fixed_size_queue.hpp" + +#include + +#include +#include +#include + +using namespace score::lcm::internal; + +class FixedSizeQueueTest : public ::testing::Test +{ + protected: + void SetUp() override + { + RecordProperty("TestType", "interface-test"); + RecordProperty("DerivationTechnique", "explorative-testing"); + } +}; + +TEST_F(FixedSizeQueueTest, StaticProperties) { + + RecordProperty("Description", "Verify that all special member functions are defined"); + + bool is_dc = std::is_default_constructible_v>; + ASSERT_TRUE(is_dc); + bool is_mc = std::is_move_constructible_v>; + ASSERT_TRUE(is_mc); + bool is_cc = std::is_copy_constructible_v>; + ASSERT_TRUE(is_cc); + bool is_ma = std::is_move_assignable_v>; + ASSERT_TRUE(is_ma); + bool is_ca = std::is_copy_assignable_v>; + ASSERT_TRUE(is_ca); + bool is_de = std::is_destructible_v>; + ASSERT_TRUE(is_de); +} + +TEST_F(FixedSizeQueueTest, ZeroCapacityAllMemberFunctionsReturns) { + + // clang-format off + RecordProperty("Description", "Verify that on zero capacity " + "full() returns true " + "empty() returns true " + "size() returns 0 " + "capacity() returns 0 " + "push returns false " + "tryPop() returns std::nullopt"); + // clang-format on + + FixedSizeQueue queue_dc; + FixedSizeQueue queue0{0}; + FixedSizeQueue zero_cap_queues[2] = {queue_dc, queue0}; + + for(FixedSizeQueue & q : zero_cap_queues) { + EXPECT_TRUE(q.full()); + EXPECT_TRUE(q.empty()); + EXPECT_EQ(q.size(), 0U); + EXPECT_EQ(q.capacity(), 0U); + EXPECT_FALSE(q.push(1)); + EXPECT_EQ(q.tryPop(), std::nullopt); + } +} + +TEST_F(FixedSizeQueueTest, CapacityFunctions) { + + // clang-format off + RecordProperty("Description", "Verify that on capacity bigger zero the capacity functions" + "full() returns false " + "empty() returns true " + "size() returns 0 " + "capacity() returns the capacity " + "initially"); + // clang-format on + + const std::size_t cap{5U}; + FixedSizeQueue queue_cap_bigger_zero{cap}; + EXPECT_FALSE(queue_cap_bigger_zero.full()); + EXPECT_TRUE(queue_cap_bigger_zero.empty()); + EXPECT_EQ(queue_cap_bigger_zero.size(), 0U); + EXPECT_EQ(queue_cap_bigger_zero.capacity(), cap); +} + +TEST_F(FixedSizeQueueTest, CapacityFunctionSizeRetrunsNumberOfElementsInTheQueue) { + + // clang-format off + RecordProperty("Description", "Verify that size() returns the number of elements in the queue, " + "that push increments the size by one, " + "and that full() returns true when the number of elements reaches the capacity."); + // clang-format on + + const std::size_t cap{5U}; + std::size_t no_elements{0U}; + FixedSizeQueue queue_cap_five{cap}; + + EXPECT_EQ(queue_cap_five.size(), no_elements); + + while(!queue_cap_five.full()) { + EXPECT_TRUE(queue_cap_five.push(1)); + ++no_elements; + EXPECT_EQ(queue_cap_five.size(), no_elements); + } + + EXPECT_EQ(queue_cap_five.capacity(), no_elements); + EXPECT_EQ(cap, no_elements); +} + +TEST_F(FixedSizeQueueTest, ModifierFunctionsPushAndTryPopImplementAFIFO) { + + // clang-format off + RecordProperty("Description", "Verify that push and tryPop implement a FIFO: " + "tryPop returns the head value, " + "tryPop decrements the size by one, " + "empty() returns true and size() returns zero once the queue is drained."); + // clang-format on + + const std::size_t cap{5U}; + FixedSizeQueue queue_cap_five{cap}; + auto elements = {11 ,22, 33, 44, 55}; + + for(auto & e : elements) { + EXPECT_TRUE(queue_cap_five.push(e)); + } + + EXPECT_EQ(queue_cap_five.size(), cap); + EXPECT_TRUE(queue_cap_five.full()); + + std::size_t no_elements{queue_cap_five.size()}; + + for(auto & e : elements) { + EXPECT_EQ(queue_cap_five.tryPop(), e); + --no_elements; + EXPECT_EQ(queue_cap_five.size(), no_elements); + } + + EXPECT_TRUE(queue_cap_five.empty()); + EXPECT_EQ(queue_cap_five.size(), 0U); +} + +TEST_F(FixedSizeQueueTest, ModifierFunctionPushReturnsFalseIfTheQueueIsFull) { + + RecordProperty("Description", "Verify that modifier function push returns false if the queue is full."); + + const std::size_t cap{5U}; + FixedSizeQueue queue_cap_five{cap}; + auto elements = {11, 22, 33, 44, 55}; + + for(auto & e : elements) { + EXPECT_TRUE(queue_cap_five.push(e)); + } + + EXPECT_TRUE(queue_cap_five.full()); + EXPECT_FALSE(queue_cap_five.push(66)); +} + +TEST_F(FixedSizeQueueTest, TailAndHeadWrap) { + + RecordProperty("Description", "Verify that tail and head wrap."); + + const std::size_t cap{3U}; + FixedSizeQueue queue_cap_three{cap}; + + auto elements = {1, 2, 3}; + for(auto & e : elements) { + EXPECT_TRUE(queue_cap_three.push(e)); + } + + EXPECT_EQ(queue_cap_three.size(), queue_cap_three.capacity()); + // {1, 2, 3} + // {H, 2, T} + + EXPECT_EQ(queue_cap_three.tryPop(), 1); + EXPECT_EQ(queue_cap_three.size(), 2U); + // { , 2, 3} + // { , H, T} + + EXPECT_TRUE(queue_cap_three.push(4)); + EXPECT_EQ(queue_cap_three.size(), queue_cap_three.capacity()); + // {4, 2, 3} + // {T, H, 3} + + EXPECT_EQ(queue_cap_three.tryPop(), 2); + EXPECT_EQ(queue_cap_three.size(), 2U); + // {4, , 3} + // {T, , H} + + + EXPECT_EQ(queue_cap_three.tryPop(), 3); + EXPECT_EQ(queue_cap_three.size(), 1U); + // {4, , } + // {TH, , } + + EXPECT_EQ(queue_cap_three.tryPop(), 4); + EXPECT_EQ(queue_cap_three.size(), 0U); +} diff --git a/score/launch_manager/src/daemon/src/common/concurrency/mpsc_bounded_queue.hpp b/score/launch_manager/src/daemon/src/common/concurrency/mpsc_bounded_queue.hpp index 89666d8a7..6ea7d5626 100644 --- a/score/launch_manager/src/daemon/src/common/concurrency/mpsc_bounded_queue.hpp +++ b/score/launch_manager/src/daemon/src/common/concurrency/mpsc_bounded_queue.hpp @@ -14,7 +14,6 @@ #ifndef MPSC_BOUNDED_QUEUE_HPP_INCLUDED #define MPSC_BOUNDED_QUEUE_HPP_INCLUDED -#include #include #include #include @@ -22,8 +21,10 @@ #include #include #include +#include #include "concurrency_error_domain.hpp" +#include "score/mw/launch_manager/common/concurrency/fixed_size_queue.hpp" #include #include @@ -37,13 +38,19 @@ namespace score::lcm::internal /// std::optional, emplaced/reset under the lock rather than pre-constructed in place. /// @warning Only one thread shall call wait()/tryPop(). Multiple threads can call push() concurrently. /// @warning push() never blocks: if the queue is full, it returns ConcurrencyErrc::kOverflow. -template +template class MpscBoundedQueue { - static_assert(Capacity > 0U, "Capacity must be at least 1"); public: - MpscBoundedQueue() = default; + MpscBoundedQueue() = delete; + + /// @brief Constructs a MpscBoundedQueue with a fixed runtime capacity. + /// @details If the provided size is 0, the capacity automatically falls back to 1. + /// @param size The desired maximum number of elements. + explicit MpscBoundedQueue(std::size_t size) : queue_((size == 0U) ? 1U : size) { + } + ~MpscBoundedQueue() { SCORE_LANGUAGE_FUTURECPP_ASSERT_MESSAGE(is_stopped(), "Call stop() and join/shut down all threads" "that might still call wait()/tryPop()/push() only then let the queue be destroyed."); @@ -78,7 +85,7 @@ class MpscBoundedQueue SCORE_LANGUAGE_FUTURECPP_ASSERT_MESSAGE(ensure_single_consumer(), "Only a single consumer thread is allowed."); const bool has_item = not_empty_cv_.wait_for(lock, timeout, [this] { - return count_ > 0U || stopped_; + return queue_.size() > 0U || stopped_; }); if (stopped_) @@ -100,17 +107,7 @@ class MpscBoundedQueue std::unique_lock lock(mutex_); SCORE_LANGUAGE_FUTURECPP_ASSERT_MESSAGE(ensure_single_consumer(), "Only a single consumer thread is allowed."); - if (count_ == 0U) - { - return std::nullopt; - } - - std::optional item = std::move(slots_[head_]); - slots_[head_].reset(); - head_ = (head_ + 1U) % Capacity; - --count_; - - return item; + return queue_.tryPop(); } /// @brief Permanently marks the queue stopped and wakes every thread currently blocked in @@ -153,34 +150,25 @@ class MpscBoundedQueue return score::cpp::make_unexpected(ConcurrencyErrc::kStopped); } - if (count_ >= Capacity) + if (!queue_.push(std::forward(item))) { return score::cpp::make_unexpected(ConcurrencyErrc::kOverflow); } - slots_[tail_].emplace(std::forward(item)); - tail_ = (tail_ + 1U) % Capacity; - ++count_; - lock.unlock(); not_empty_cv_.notify_one(); return {}; } - mutable std::mutex mutex_; + mutable std::mutex mutex_{}; /// @brief Signaled by push()/stop(); waited on by wait(). - std::condition_variable not_empty_cv_; - std::array, Capacity> slots_{}; - /// @brief Index of the oldest item, valid only when count_ > 0. - std::size_t head_{0}; - /// @brief Index where the next push lands. - std::size_t tail_{0}; - /// @brief Number of occupied slots, guarded by mutex_. - std::size_t count_{0}; - /// @brief Guarded by mutex_. + std::condition_variable not_empty_cv_{}; + /// @brief Queue holding the elements. + FixedSizeQueue queue_; + /// @brief True if the queue has been stopped, false otherwise. bool stopped_{false}; - /// @brief Consumer thread id - std::thread::id consumer_thread_id_; + /// @brief Consumer thread id. + std::thread::id consumer_thread_id_{}; }; } // namespace score::lcm::internal diff --git a/score/launch_manager/src/daemon/src/common/concurrency/mpsc_bounded_queue_test.cpp b/score/launch_manager/src/daemon/src/common/concurrency/mpsc_bounded_queue_test.cpp index b9fa87fc8..c4815ec78 100644 --- a/score/launch_manager/src/daemon/src/common/concurrency/mpsc_bounded_queue_test.cpp +++ b/score/launch_manager/src/daemon/src/common/concurrency/mpsc_bounded_queue_test.cpp @@ -32,7 +32,7 @@ class MpscBoundedQueueTest_Basic : public ::testing::Test queue_.stop(); } - MpscBoundedQueue queue_; + MpscBoundedQueue queue_{8U}; }; TEST_F(MpscBoundedQueueTest_Basic, PushAndPopSingleItem) @@ -88,7 +88,7 @@ TEST_F(MpscBoundedQueueTest_Basic, RvaluePushWorksWithMoveOnlyType) }; static_assert(!std::is_default_constructible_v); - MpscBoundedQueue queue; + MpscBoundedQueue queue{4U}; ASSERT_TRUE(queue.push(MoveOnly{std::make_unique(99)}).has_value()); auto result = queue.tryPop(); @@ -120,7 +120,7 @@ class MpscBoundedQueueTest_Capacity : public ::testing::Test } static constexpr std::size_t kCapacity = 4U; - MpscBoundedQueue queue_; + MpscBoundedQueue queue_{kCapacity}; }; TEST_F(MpscBoundedQueueTest_Capacity, FillsExactlyToCapacity) @@ -194,12 +194,22 @@ TEST_F(MpscBoundedQueueTest_Capacity, RepeatedFillDrainCyclesWrapRingBufferCorre EXPECT_FALSE(queue_.tryPop().has_value()); } +TEST(MpscBoundedQueueTest_MinimalCapacity, CapacityOfZeroIsOne) +{ + RecordProperty("Description", + "Verify the smallest legal capacity of one is guaranteed."); + MpscBoundedQueue queue0{0U}; + EXPECT_TRUE(queue0.push(1).has_value()) << "the smallest legal capacity of one shall be guaranteed"; + EXPECT_FALSE(queue0.push(2).has_value()) << "only one element is allowed"; + queue0.stop(); +} + TEST(MpscBoundedQueueTest_MinimalCapacity, CapacityOfOneAllowsSingleItemAtATime) { RecordProperty("Description", "Verify the smallest legal capacity (1) behaves correctly: a single push succeeds, a second " "push overflows until the first item is drained, matching general boundary behavior."); - MpscBoundedQueue queue; + MpscBoundedQueue queue{1U}; EXPECT_TRUE(queue.push(1).has_value()); auto blocked = queue.push(2); @@ -226,8 +236,8 @@ class MpscBoundedQueueTest_Blocking : public ::testing::Test queue8_.stop(); } - MpscBoundedQueue queue4_; - MpscBoundedQueue queue8_; + MpscBoundedQueue queue4_{4U}; + MpscBoundedQueue queue8_{8U}; }; TEST_F(MpscBoundedQueueTest_Blocking, WaitReturnsTimeoutWhenEmpty) @@ -272,8 +282,8 @@ class MpscBoundedQueueTest_Stop : public ::testing::Test queue8_.stop(); } - MpscBoundedQueue queue4_; - MpscBoundedQueue queue8_; + MpscBoundedQueue queue4_{4U}; + MpscBoundedQueue queue8_{8U}; }; TEST_F(MpscBoundedQueueTest_Stop, PushReturnsStoppedAfterStop) @@ -349,7 +359,7 @@ class MpscBoundedQueueTest_MultiProducer : public ::testing::Test static constexpr std::size_t kNumProducers = 6U; static constexpr int kTotalItems = static_cast(kItemsPerProducer * kNumProducers); - MpscBoundedQueue queue_; + MpscBoundedQueue queue_{kCapacity}; }; TEST_F(MpscBoundedQueueTest_MultiProducer, AllItemsFromConcurrentProducersDeliveredExactlyOnce) @@ -436,16 +446,16 @@ TEST_F(MpscBoundedQueueTest_MultiProducer, AllItemsFromConcurrentProducersDelive TEST(MpscBoundedQueueTest, ToCallStopBeforeDestructionIsMandatory) { RecordProperty("Description", "Verfiy that calling stop before destruction is mandatory."); - using IntQueue = MpscBoundedQueue; - EXPECT_DEATH({IntQueue{};}, ""); + using IntQueue = MpscBoundedQueue; + EXPECT_DEATH({IntQueue{4};}, ""); } TEST(MpscBoundedQueueTest, OnlyOneConsumerThreadIsAllowed) { RecordProperty("Description", "Verfiy that only one consumer thread is allowed."); - using IntQueue = MpscBoundedQueue; + using IntQueue = MpscBoundedQueue; EXPECT_DEATH({ - IntQueue queue; + IntQueue queue{4}; std::thread consumer([&] { static_cast(queue.wait(std::chrono::milliseconds{5000})); });