diff --git a/score/launch_manager/src/daemon/src/common/concurrency/BUILD b/score/launch_manager/src/daemon/src/common/concurrency/BUILD index c17d3a7b6..a38b25973 100644 --- a/score/launch_manager/src/daemon/src/common/concurrency/BUILD +++ b/score/launch_manager/src/daemon/src/common/concurrency/BUILD @@ -58,6 +58,20 @@ cc_binary( ], ) +cc_library( + name = "mpsc_bounded_queue", + hdrs = [ + "concurrency_error_domain.hpp", + "mpsc_bounded_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 = "mpmc_concurrent_queue_test", srcs = ["mpmc_concurrent_queue_test.cpp"], 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 new file mode 100644 index 000000000..3a0c5df8e --- /dev/null +++ b/score/launch_manager/src/daemon/src/common/concurrency/mpsc_bounded_queue.hpp @@ -0,0 +1,161 @@ +/******************************************************************************** + * 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 MPSC_BOUNDED_QUEUE_HPP_INCLUDED +#define MPSC_BOUNDED_QUEUE_HPP_INCLUDED + +#include +#include +#include +#include +#include +#include +#include + +#include "concurrency_error_domain.hpp" +#include + +namespace score::lcm::internal +{ + +/// @brief Fixed-capacity queue for the multi-producer / single-consumer case, +/// built on a plain std::mutex + std::condition_variable rather than a lock-free scheme. +/// @details Unlike MPMCConcurrentQueue, T does not need to be default-constructible: slots are +/// 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 +class MpscBoundedQueue +{ + static_assert(Capacity > 0U, "Capacity must be at least 1"); + + public: + MpscBoundedQueue() = default; + ~MpscBoundedQueue() = default; + + MpscBoundedQueue(const MpscBoundedQueue&) = delete; + MpscBoundedQueue& operator=(const MpscBoundedQueue&) = delete; + MpscBoundedQueue(MpscBoundedQueue&&) = delete; + MpscBoundedQueue& operator=(MpscBoundedQueue&&) = delete; + + /// @brief Enqueues an item. + /// @return blank on success; ConcurrencyErrc::kOverflow if the queue is full, or + /// ConcurrencyErrc::kStopped if the queue has been stopped. + [[nodiscard]] score::cpp::expected_blank push(T&& item) + { + return push_impl(std::move(item)); + } + + [[nodiscard]] score::cpp::expected_blank push(const T& item) + { + return push_impl(item); + } + + /// @brief Waits up to `timeout` for at least one item to become available. + /// @param timeout Maximum time to wait. + /// @return blank if an item is available (caller should drain via tryPop()); + /// ConcurrencyErrc::kTimeout if the timeout elapsed with none available, or + /// ConcurrencyErrc::kStopped if the queue has been stopped. + [[nodiscard]] score::cpp::expected_blank wait(std::chrono::milliseconds timeout) + { + std::unique_lock lock(mutex_); + const bool has_item = not_empty_cv_.wait_for(lock, timeout, [this] { + return count_ > 0U || stopped_; + }); + + if (stopped_) + { + return score::cpp::make_unexpected(ConcurrencyErrc::kStopped); + } + if (!has_item) + { + return score::cpp::make_unexpected(ConcurrencyErrc::kTimeout); + } + return {}; + } + + /// @brief Non-blocking pop. Never waits, regardless of whether the queue has been stopped, so + /// that anything still queued at shutdown time can still be drained. + /// @return The next item if any is available right now, or std::nullopt if empty. + std::optional tryPop() + { + std::unique_lock lock(mutex_); + + if (count_ == 0U) + { + return std::nullopt; + } + + std::optional item = std::move(slots_[head_]); + slots_[head_].reset(); + head_ = (head_ + 1U) % Capacity; + --count_; + + return item; + } + + /// @brief Permanently marks the queue stopped and wakes every thread currently blocked in + /// wait(). After this call, push() and wait() always return + /// ConcurrencyErrc::kStopped. tryPop() is unaffected and continues to drain any + /// queued items. + void stop() + { + { + std::lock_guard lock(mutex_); + stopped_ = true; + } + not_empty_cv_.notify_all(); + } + + private: + template + [[nodiscard]] score::cpp::expected_blank push_impl(U&& item) + { + std::unique_lock lock(mutex_); + + if (stopped_) + { + return score::cpp::make_unexpected(ConcurrencyErrc::kStopped); + } + + if (count_ >= Capacity) + { + 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_; + /// @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_. + bool stopped_{false}; +}; + +} // namespace score::lcm::internal + +#endif // MPSC_BOUNDED_QUEUE_HPP_INCLUDED