Skip to content

Adding new mpmc queue#187

Open
MaciejKaszynski wants to merge 10 commits intoeclipse-score:mainfrom
etas-contrib:new-queue
Open

Adding new mpmc queue#187
MaciejKaszynski wants to merge 10 commits intoeclipse-score:mainfrom
etas-contrib:new-queue

Conversation

@MaciejKaszynski
Copy link
Copy Markdown
Contributor

  • Adding new queue

@github-actions
Copy link
Copy Markdown

github-actions Bot commented Apr 28, 2026

License Check Results

🚀 The license check job ran with the Bazel command:

bazel run --lockfile_mode=error //:license-check

Status: ⚠️ Needs Review

Click to expand output
[License Check Output]
Extracting Bazel installation...
Starting local Bazel server (8.4.2) and connecting to it...
INFO: Invocation ID: e63234f1-fa85-4494-b3f6-29c5174568c9
Computing main repo mapping: 
Computing main repo mapping: 
Loading: 
Loading: 0 packages loaded
Loading: 0 packages loaded
Loading: 0 packages loaded
    currently loading: 
Loading: 0 packages loaded
    currently loading: 
Analyzing: target //:license-check (1 packages loaded, 0 targets configured)
Analyzing: target //:license-check (1 packages loaded, 0 targets configured)

Analyzing: target //:license-check (27 packages loaded, 10 targets configured)

Analyzing: target //:license-check (85 packages loaded, 10 targets configured)

Analyzing: target //:license-check (137 packages loaded, 2708 targets configured)

Analyzing: target //:license-check (149 packages loaded, 5380 targets configured)

Analyzing: target //:license-check (154 packages loaded, 5429 targets configured)

Analyzing: target //:license-check (156 packages loaded, 8076 targets configured)

Analyzing: target //:license-check (161 packages loaded, 10106 targets configured)

INFO: Analyzed target //:license-check (162 packages loaded, 10232 targets configured).
[4 / 16] Creating runfiles tree bazel-out/k8-fastbuild/bin/license.check.license_check.runfiles; 0s local ... (2 actions, 1 running)
[12 / 16] JavaToolchainCompileClasses external/rules_java+/toolchains/platformclasspath_classes; 0s disk-cache, processwrapper-sandbox ... (2 actions running)
[15 / 16] [Prepa] Building license.check.license_check.jar ()
INFO: Found 1 target...
Target //:license.check.license_check up-to-date:
  bazel-bin/license.check.license_check
  bazel-bin/license.check.license_check.jar
INFO: Elapsed time: 21.279s, Critical Path: 2.37s
INFO: 16 processes: 12 internal, 3 processwrapper-sandbox, 1 worker.
INFO: Build completed successfully, 16 total actions
INFO: Running command line: bazel-bin/license.check.license_check ./formatted.txt <args omitted>
usage: org.eclipse.dash.licenses.cli.Main [-batch <int>] [-cd <url>]
       [-confidence <int>] [-ef <url>] [-excludeSources <sources>] [-help] [-lic
       <url>] [-project <shortname>] [-repo <url>] [-review] [-summary <file>]
       [-timeout <seconds>] [-token <token>]

@github-actions
Copy link
Copy Markdown

The created documentation from the pull request is available at: docu-html

Copy link
Copy Markdown

Copilot AI left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

This PR replaces the existing JobQueue with a new MPMCConcurrentQueue implementation and wires it into the Launch Manager daemon’s process-group worker execution flow.

Changes:

  • Introduces MPMCConcurrentQueue (MPMC ring buffer + semaphore blocking) plus Helgrind annotations and Bazel targets/tests.
  • Updates WorkerThread, ProcessGroupManager, Graph, and ProcessInfoNode to enqueue/dequeue work via the new queue API (push/pop/stop).
  • Removes the old JobQueue implementation and adjusts Bazel coverage/test filtering and OSAL source selection.

Reviewed changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 10 comments.

Show a summary per file
File Description
src/launch_manager_daemon/src/process_group_manager/workerthread.hpp Switches worker thread pool to use MPMCConcurrentQueue and adds stop() API.
src/launch_manager_daemon/src/process_group_manager/workerthread.cpp Implements new pop-driven worker loop and queue stop behavior.
src/launch_manager_daemon/src/process_group_manager/processinfonode.cpp Migrates successor job enqueueing to push(..., kMaxQueueDelay).
src/launch_manager_daemon/src/process_group_manager/processgroupmanager.hpp Replaces JobQueue type with WorkerQueue alias over MPMCConcurrentQueue.
src/launch_manager_daemon/src/process_group_manager/processgroupmanager.cpp Constructs the new queue and updates shutdown timeout behavior to stop workers.
src/launch_manager_daemon/src/process_group_manager/jobqueue.hpp Removes old job queue header.
src/launch_manager_daemon/src/process_group_manager/jobqueue.cpp Removes old job queue implementation.
src/launch_manager_daemon/src/process_group_manager/graph.cpp Updates node queuing to push(..., kMaxQueueDelay) and logging.
src/launch_manager_daemon/src/concurrency/mpmc_concurrent_queue.hpp Adds the new concurrent queue implementation.
src/launch_manager_daemon/src/concurrency/helgrind_annotations.hpp Adds optional Helgrind annotation macros.
src/launch_manager_daemon/src/concurrency/mpmc_concurrent_queue_test.cpp Adds unit tests for queue correctness and concurrency behaviors.
src/launch_manager_daemon/src/concurrency/BUILD Adds Bazel targets for the queue library and tests (including helgrind/tsan variants).
src/launch_manager_daemon/health_monitor_lib/src/score/lcm/saf/supervision/ProcessStateTracker.cpp Minor iterator declaration tweak.
src/launch_manager_daemon/common/BUILD Refactors OSAL source globs to include posix + OS-specific sources.
src/launch_manager_daemon/BUILD Adds dependency on the new concurrency queue library.
.bazelrc Excludes no-coverage tagged tests from coverage runs.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Comment thread src/launch_manager_daemon/src/process_group_manager/processinfonode.cpp Outdated
Comment thread src/launch_manager_daemon/src/concurrency/BUILD Outdated
Comment thread src/launch_manager_daemon/src/process_group_manager/workerthread.cpp Outdated
Comment thread src/launch_manager_daemon/src/concurrency/mpmc_concurrent_queue.hpp Outdated
Comment thread src/launch_manager_daemon/src/process_group_manager/graph.cpp Outdated
Comment thread src/launch_manager_daemon/src/process_group_manager/processinfonode.cpp Outdated
Comment thread src/launch_manager_daemon/src/process_group_manager/workerthread.cpp Outdated
Comment thread src/launch_manager_daemon/src/process_group_manager/graph.cpp Outdated
Comment thread src/launch_manager_daemon/src/process_group_manager/processinfonode.cpp Outdated
Comment thread src/launch_manager_daemon/src/process_group_manager/processinfonode.cpp Outdated
Copy link
Copy Markdown

Copilot AI left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Copilot reviewed 19 out of 19 changed files in this pull request and generated 5 comments.

Comments suppressed due to low confidence (1)

src/launch_manager_daemon/src/process_group_manager/processinfonode.cpp:195

  • In queueTerminationSuccessorJobs(), the retry loop keeps running as long as the graph is in transition, but it doesn't handle non-timeout failures from push() (e.g., ConcurrencyErrc::kStopped / kOsError). If the worker queue has been stopped, push() will fail immediately and this loop can spin indefinitely. Handle the error code (retry on timeout, break/abort on stopped/OS error) to avoid a potential hang during shutdown/timeout paths.
        if (successor_node->is_included_ && successor_node->dependencies_ > 0U && --successor_node->dependencies_ == 0U)
        {
            while (graph_->getState() == GraphState::kInTransition)
            {
                if (graph_->getProcessGroupManager()->getWorkerJobs()->push(successor_node, kMaxQueueDelay))
                {
                    graph_->markNodeInFlight();
                    break;
                }
            }

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Comment thread src/launch_manager_daemon/src/concurrency/mpmc_concurrent_queue.hpp
Comment on lines +131 to +173
[[nodiscard]] score::Result<void> stop() noexcept
{
m_stopped.store(true, std::memory_order_relaxed);

// signal to consumers and publishers to wakeup
if (m_items.post() != osal::OsalReturnType::kSuccess)
{
return score::MakeUnexpected(ConcurrencyErrc::kOsError);
}

if (m_spaces.post() != osal::OsalReturnType::kSuccess)
{
return score::MakeUnexpected(ConcurrencyErrc::kOsError);
}

return {};
}

/// @brief Blocks until an item is available or stop() is called.
/// @details Consumers claim slots via fetch_add on m_head and sleep
/// inside m_items.wait() when the queue is empty.
/// When stopped returns std::nullopt.
/// @param timeout Maximum time to wait for an item. Zero means wait forever.
/// @return The next item, or error.
[[nodiscard]] score::Result<T> pop(std::chrono::milliseconds timeout = std::chrono::milliseconds{0})
{
const auto wait_result =
(timeout == std::chrono::milliseconds{0}) ? m_items.wait() : m_items.timedWait(timeout);

if(wait_result == osal::OsalReturnType::kTimeout)
{
return score::MakeUnexpected(ConcurrencyErrc::kTimeout);
}
else if (wait_result != osal::OsalReturnType::kSuccess)
{
return score::MakeUnexpected(ConcurrencyErrc::kOsError);
}

if (m_stopped.load(std::memory_order_relaxed))
{
static_cast<void>(m_items.post());
return score::MakeUnexpected(ConcurrencyErrc::kStopped);
}
Copy link

Copilot AI Apr 30, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

stop() sets m_stopped using memory_order_relaxed, and push()/pop() also read it with memory_order_relaxed. Because the semaphore wait/post calls are outside the C++ memory model, relaxed ordering can allow the compiler/CPU to reorder such that a woken thread still observes m_stopped == false and continues consuming/producing after stop(). Use at least release on the store in stop() and acquire on the loads in push()/pop() (or an explicit fence) to make the stop flag reliably observable before/after the wakeups.

Copilot uses AI. Check for mistakes.
Comment thread src/launch_manager_daemon/src/concurrency/mpmc_concurrent_queue.hpp
Comment thread src/launch_manager_daemon/src/process_group_manager/workerthread.cpp Outdated
Copy link
Copy Markdown

Copilot AI left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Copilot reviewed 19 out of 19 changed files in this pull request and generated 3 comments.

Comments suppressed due to low confidence (1)

src/launch_manager_daemon/src/process_group_manager/processinfonode.cpp:195

  • In queueTerminationSuccessorJobs(), the while (graph_->getState() == GraphState::kInTransition) loop retries push() on any failure. If the worker queue has been stopped (push returns ConcurrencyErrc::kStopped) or another non-timeout error occurs, push() can fail immediately and this becomes a tight infinite loop, potentially deadlocking shutdown/abort paths. Handle the push() result (e.g., continue only on kTimeout, and break/return on kStopped or other errors, optionally logging once).
        if (successor_node->is_included_ && successor_node->dependencies_ > 0U && --successor_node->dependencies_ == 0U)
        {
            while (graph_->getState() == GraphState::kInTransition)
            {
                if (graph_->getProcessGroupManager()->getWorkerJobs()->push(successor_node, kMaxQueueDelay))
                {
                    graph_->markNodeInFlight();
                    break;
                }
            }

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Comment thread src/launch_manager_daemon/src/concurrency/BUILD
Comment on lines +36 to +55
/// @brief Lock-free MPMC ring buffer with semaphore-based blocking.
/// Producers and consumers each atomically claim independent slots via
/// fetch_add, so multiple producers and multiple consumers run concurrently.
/// @warning T must be default-constructible.
template <typename T, std::size_t Capacity>
class MPMCConcurrentQueue
{
static_assert(Capacity > 0U, "Capacity must be at least 1");

static_assert(Capacity <= std::numeric_limits<std::uint32_t>::max(),
"Capacity exceeds uint32_t range used by the semaphore");

static_assert(std::is_default_constructible_v<T>, "T must be default-constructible for in-place slot storage");

static_assert(std::is_nothrow_destructible_v<T>,
"T must be nothrow-destructible to allow consume_slot to be noexcept");

static_assert(std::is_nothrow_move_constructible_v<T>,
"T must be nothrow-move-constructible to wrap into std::optional in pop()");

Copy link

Copilot AI Apr 30, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The template constraints/documentation say only that T must be default-constructible (and nothrow-move-constructible), but push_impl() assigns into an already-constructed slot.item (slot.item = ...). That also requires T to be move-assignable (and copy-assignable for the push(const T&) overload). Consider either tightening the static_asserts / docs to reflect the real requirements, or switching slot storage to placement-new / std::optional<T> so non-assignable-but-move-constructible types work as well.

Copilot uses AI. Check for mistakes.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants