Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ cc_library(
"//score/launch_manager/src/daemon/src/configuration:config",
"//score/launch_manager/src/daemon/src/osal:ipc_comms",
"//score/launch_manager/src/daemon/src/osal:wait_for_file",
"//score/launch_manager/src/daemon/src/process_group_manager/details:completion_slot",
"//score/launch_manager/src/daemon/src/process_group_manager/details:component_event",
"//score/launch_manager/src/daemon/src/process_group_manager/details:graph",
"//score/launch_manager/src/daemon/src/process_group_manager/details:os_handler",
Expand All @@ -65,7 +66,6 @@ cc_library(
"//score/launch_manager/src/daemon/src/recovery_client",
"//score/launch_manager/src/daemon/src/watchdog:i_watchdog_if",
"//score/launch_manager/src/lm_control",
"@score_baselibs//score/concurrency/future",
"@score_baselibs//score/language/futurecpp",
],
)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -88,16 +88,29 @@ cc_library(
],
)

cc_library(
name = "completion_slot",
hdrs = ["completion_slot.hpp"],
include_prefix = "score/mw/launch_manager/process_group_manager/details",
strip_include_prefix = "/score/launch_manager/src/daemon/src/process_group_manager/details",
visibility = ["//score/launch_manager/src/daemon/src/process_group_manager:__subpackages__"],
deps = [
"@score_baselibs//score/concurrency:notification",
"@score_baselibs//score/language/futurecpp",
],
)

cc_library(
name = "component_event",
hdrs = ["component_event.hpp"],
include_prefix = "score/mw/launch_manager/process_group_manager/details",
strip_include_prefix = "/score/launch_manager/src/daemon/src/process_group_manager/details",
visibility = ["//score/launch_manager/src/daemon/src/process_group_manager:__pkg__"],
deps = [
":completion_slot",
":icomponent",
"//score/launch_manager/src/daemon/src/common:identifier_hash",
"@score_baselibs//score/concurrency/future",
"@score_baselibs//score/result",
],
)

Expand Down Expand Up @@ -128,6 +141,17 @@ cc_library(
],
)

lm_cc_test(
name = "completion_slot_UT",
srcs = ["completion_slot_UT.cpp"],
deps = [
":completion_slot",
"//score/launch_manager/src/daemon/src/common:identifier_hash",
"@googletest//:gtest_main",
"@score_baselibs//score/result",
],
)

lm_cc_test(
name = "component_event_queue_UT",
srcs = ["component_event_queue_UT.cpp"],
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
/********************************************************************************
* 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 SCORE_LCM_COMPLETION_SLOT_HPP_INCLUDED
#define SCORE_LCM_COMPLETION_SLOT_HPP_INCLUDED

#include "score/concurrency/notification.h"
#include "score/stop_token.hpp"

#include <optional>
#include <utility>

namespace score::mw::lifecycle::internal
{

/// @brief A non-allocating, single-producer, single-consumer one-shot completion slot.
///
/// Designed to be allocated on the caller's stack frame. The caller passes a non-owning
/// pointer to the slot via an event to the main event loop thread and blocks unconditionally
/// in wait() until the consumer completes it. This guarantees the slot remains valid on the
/// stack for the duration of the operation, avoiding dynamic memory allocation without
/// exposing unsafe cancellation that could lead to a use-after-free.
template <typename T>
class CompletionSlot final
{
public:
CompletionSlot() = default;
~CompletionSlot() = default;
CompletionSlot(const CompletionSlot&) = delete;
CompletionSlot& operator=(const CompletionSlot&) = delete;
CompletionSlot(CompletionSlot&&) = delete;
CompletionSlot& operator=(CompletionSlot&&) = delete;

/// @brief Publishes the result and wakes the waiting thread. Called once, typically by the main thread.
void complete(T value)
{
value_ = std::move(value);
done_.notify();
}

/// @brief Blocks unconditionally until complete() has run. Called once, by the owning thread.
/// @return The completed value.
[[nodiscard]] T wait()
{
static_cast<void>(done_.waitWithAbort(score::cpp::stop_token{}));
return std::move(*value_);
}

private:
/// @brief Inline storage for the result; no dynamic memory allocation.
std::optional<T> value_{};

/// @brief Notification held by value (mutex + cv + flag). No dynamic memory allocation.
score::concurrency::Notification done_{};
};

} // namespace score::mw::lifecycle::internal

#endif // SCORE_LCM_COMPLETION_SLOT_HPP_INCLUDED
Original file line number Diff line number Diff line change
@@ -0,0 +1,156 @@
/********************************************************************************
* 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/process_group_manager/details/completion_slot.hpp"

#include <gtest/gtest.h>

#include <atomic>
#include <chrono>
#include <cstdlib>
#include <new>
#include <thread>

#include "score/mw/launch_manager/common/identifier_hash.hpp"
#include "score/result/result.h"

namespace
{

thread_local bool g_tracking_enabled = false;
thread_local std::size_t g_tracked_allocations = 0;

} // namespace

void* operator new(std::size_t size)
{
if (g_tracking_enabled)
{
++g_tracked_allocations;
}
void* ptr = std::malloc(size);
if (ptr == nullptr)
{
throw std::bad_alloc();
}
return ptr;
}

void operator delete(void* ptr) noexcept
{
std::free(ptr);
}

void operator delete(void* ptr, std::size_t) noexcept
{
std::free(ptr);
}

namespace score::mw::lifecycle::internal
{
namespace
{

class CompletionSlotTest : public ::testing::Test
{
protected:
void SetUp() override
{
RecordProperty("TestType", "unit-test");
RecordProperty("DerivationTechnique", "requirements-based");
}
};

TEST_F(CompletionSlotTest, SynchronousCompleteThenWaitReturnsValue)
{
RecordProperty("Description", "Verify CompletionSlot returns the published value synchronously.");

CompletionSlot<int> slot;
slot.complete(42);

int result = slot.wait();
EXPECT_EQ(result, 42);
}

TEST_F(CompletionSlotTest, SynchronousCompleteResultVoid)
{
RecordProperty("Description", "Verify CompletionSlot works with score::Result<void>.");

CompletionSlot<score::Result<void>> slot;
slot.complete(score::Result<void>{});

auto result = slot.wait();
EXPECT_TRUE(result.has_value());
}

TEST_F(CompletionSlotTest, AsynchronousCompleteFromWorkerThread)
{
RecordProperty("Description", "Verify CompletionSlot unblocks waiting thread when completed asynchronously.");

CompletionSlot<score::Result<IdentifierHash>> slot;
const IdentifierHash expected_hash{"TargetRunState"};

std::thread worker([&slot, expected_hash]() {
std::this_thread::sleep_for(std::chrono::milliseconds{10});
slot.complete(score::Result<IdentifierHash>{expected_hash});
});

auto result = slot.wait();
worker.join();

ASSERT_TRUE(result.has_value());
EXPECT_EQ(result.value(), expected_hash);
}

TEST_F(CompletionSlotTest, ZeroAllocationInCompleteAndWait)
{
RecordProperty(
"Description", "Verify CompletionSlot complete() and wait() perform zero dynamic memory allocations.");

const IdentifierHash test_hash{"Running"};

g_tracked_allocations = 0;
g_tracking_enabled = true;

{
CompletionSlot<score::Result<IdentifierHash>> slot;
slot.complete(score::Result<IdentifierHash>{test_hash});
auto result = slot.wait();
ASSERT_TRUE(result.has_value());
EXPECT_EQ(result.value(), test_hash);
}

g_tracking_enabled = false;
EXPECT_EQ(g_tracked_allocations, 0U);
}

TEST_F(CompletionSlotTest, ZeroAllocationInResultVoidCompleteAndWait)
{
RecordProperty("Description", "Verify CompletionSlot with Result<void> performs zero dynamic memory allocations.");

g_tracked_allocations = 0;
g_tracking_enabled = true;

{
CompletionSlot<score::Result<void>> slot;
slot.complete(score::Result<void>{});
auto result = slot.wait();
EXPECT_TRUE(result.has_value());
}

g_tracking_enabled = false;
EXPECT_EQ(g_tracked_allocations, 0U);
}

} // namespace
} // namespace score::mw::lifecycle::internal
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,10 @@
#include <cstdint>
#include <variant>

#include "score/concurrency/future/interruptible_promise.h"
#include "score/mw/launch_manager/common/identifier_hash.hpp"
#include "score/mw/launch_manager/process_group_manager/details/completion_slot.hpp"
#include "score/mw/launch_manager/process_group_manager/details/icomponent.hpp"
#include "score/result/result.h"

namespace score::mw::lifecycle::internal
{
Expand Down Expand Up @@ -73,14 +74,14 @@ struct [[nodiscard]] SupervisionFailure
/// @brief The state manager wants to know which Run Target is currently active.
struct [[nodiscard]] GetActiveRunTarget
{
concurrency::InterruptiblePromise<Result<IdentifierHash>> promise;
CompletionSlot<Result<IdentifierHash>>* completion{nullptr};
};

/// @brief The state manager wants to change to a different Run Target.
struct [[nodiscard]] SetRequestedRunTarget
{
IdentifierHash run_target;
concurrency::InterruptiblePromise<Result<void>> promise;
CompletionSlot<Result<void>>* completion{nullptr};
};

/// @brief A graph-relevant state change. There is only ever a single graph, so no process-group
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -51,9 +51,12 @@ class ComponentEventQueue final : public IComponentEventPublisherConsumer
[[nodiscard]] bool push(ComponentEvent&& event) override
{
auto result = queue_.push(std::move(event));
if (!result.has_value() && result.error() == mw::lifecycle::internal::ConcurrencyErrc::kOverflow)
if (!result.has_value())
{
overflow_.store(true, std::memory_order_release);
if (result.error() == mw::lifecycle::internal::ConcurrencyErrc::kOverflow)
{
overflow_.store(true, std::memory_order_release);
}
return false;
}
return true;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,38 @@ TEST_F(ComponentEventQueueTest, GetNextEventReturnsSupervisionFailureWithPayload
EXPECT_EQ(failure.process_identifier, process_identifier);
}

TEST_F(ComponentEventQueueTest, GetNextEventReturnsGetActiveRunTargetWithCompletionSlot)
{
RecordProperty(
"Description", "Verify a pushed GetActiveRunTarget event is returned with its CompletionSlot pointer intact.");
CompletionSlot<Result<IdentifierHash>> slot;
EXPECT_TRUE(queue_.push(GetActiveRunTarget{completion : &slot}));

auto event = queue_.getNextEvent();
ASSERT_TRUE(event.has_value());
ASSERT_TRUE(std::holds_alternative<GetActiveRunTarget>(*event));
const auto& get_active = std::get<GetActiveRunTarget>(*event);
EXPECT_EQ(get_active.completion, &slot);
}

TEST_F(ComponentEventQueueTest, GetNextEventReturnsSetRequestedRunTargetWithPayloadAndCompletionSlot)
{
RecordProperty(
"Description",
"Verify a pushed SetRequestedRunTarget event is returned with its run_target and CompletionSlot pointer "
"intact.");
CompletionSlot<Result<void>> slot;
const IdentifierHash target{"RunTargetA"};
EXPECT_TRUE(queue_.push(SetRequestedRunTarget{run_target : target, completion : &slot}));

auto event = queue_.getNextEvent();
ASSERT_TRUE(event.has_value());
ASSERT_TRUE(std::holds_alternative<SetRequestedRunTarget>(*event));
const auto& set_target = std::get<SetRequestedRunTarget>(*event);
EXPECT_EQ(set_target.run_target, target);
EXPECT_EQ(set_target.completion, &slot);
}

TEST_F(ComponentEventQueueTest, GetOverflowStaysFalseUnderNormalUsage)
{
RecordProperty("Description", "Verify getOverflow() stays false when events are pushed and drained normally.");
Expand Down Expand Up @@ -132,4 +164,11 @@ TEST_F(ComponentEventQueueTest, GetNextEventStillDrainsQueuedEventsAfterStop)
EXPECT_FALSE(queue_.getNextEvent().has_value());
}

TEST_F(ComponentEventQueueTest, PushReturnsFalseAfterStop)
{
RecordProperty("Description", "Verify push returns false immediately if the queue has been stopped.");
queue_.stop();
EXPECT_FALSE(queue_.push(ActivationSuccessful{IdentifierHash{"process"}}));
}

} // namespace score::mw::lifecycle::internal
Loading
Loading