Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
26 commits
Select commit Hold shift + click to select a range
5c54c1b
ProcessGroupMonitor Refactoring
WilliamRoebuck Jun 5, 2026
73d61f1
Fix build and unit tests
NicolasFussberger Jul 2, 2026
6182f47
Initial fixes after rebase
WilliamRoebuck Jul 29, 2026
0324bc4
Revert test changes
WilliamRoebuck Jul 30, 2026
6bcb709
Cleanup
WilliamRoebuck Jul 30, 2026
3308ae8
Introduce interface for SafeProcessMap
NicolasFussberger Jul 30, 2026
d6df09b
Restore force termination method
WilliamRoebuck Jul 30, 2026
8f42c43
Setup mocks
NicolasFussberger Jul 30, 2026
b17e9c5
ProcessInfoNode unit tests
NicolasFussberger Jul 30, 2026
15d8e99
Remove stop_token usage
NicolasFussberger Jul 31, 2026
7b3372b
Include safeprocessmap header
NicolasFussberger Jul 31, 2026
a9382a4
Assert more process state transitions
NicolasFussberger Jul 31, 2026
8a17836
Add interfaces to Graph constructor for UTs
WilliamRoebuck Jul 30, 2026
964105c
Initial graph UT
WilliamRoebuck Jul 31, 2026
3db0030
Further changes & fixes for UT
WilliamRoebuck Aug 3, 2026
6d16490
Remove some extra code, more UTs
WilliamRoebuck Aug 3, 2026
7d85000
Fixes, UTs for off transition, UTs for handle event
WilliamRoebuck Aug 3, 2026
d3ca9b8
Cancel & util tests
WilliamRoebuck Aug 3, 2026
fb0af73
Fixes after rebase
WilliamRoebuck Aug 4, 2026
4a8d42f
Add job skipped event
WilliamRoebuck Aug 4, 2026
86a8437
Add missing atomic load
WilliamRoebuck Aug 4, 2026
544250e
Fix flaky failure
WilliamRoebuck Aug 4, 2026
e855e15
Cleanup
WilliamRoebuck Aug 5, 2026
6cbd7fe
Add doc, add failed state checks
WilliamRoebuck Aug 5, 2026
40a2e6e
Update some asserts
WilliamRoebuck Aug 5, 2026
1b2f198
Merge branch 'main' into feature/process-mon-refactor-2
WilliamRoebuck Aug 5, 2026
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
4 changes: 4 additions & 0 deletions score/launch_manager/docs/user_guide/configuration.rst
Original file line number Diff line number Diff line change
Expand Up @@ -357,6 +357,10 @@ run_targets (object, optional)
* **Description:** Defines an individual **Run Target's** configuration, specifying the components and dependencies that constitute a particular operational mode.
* **Reference:** This property refers to the ``run_target`` reusable type defined in this schema.

The name "Off" is reserved for a special **Run Target** that represents the system's shutdown state.
If a **Run Target** with this name is configured explicitly, the **Launch Manager** will transition to this **Run Target** when receiving a SIGTERM signal.
If not configured explicitly, the **Launch Manager** will automatically create a default **Run Target** named "Off" with no dependencies.

.. _lm_conf_initial_run_target_string_required_:

initial_run_target (string, required)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,8 @@ class ProcessState : public saf::common::Observable<ProcessState>
starting = static_cast<uint8_t>(score::lcm::ProcessState::kStarting),
running = static_cast<uint8_t>(score::lcm::ProcessState::kRunning),
sigterm = static_cast<uint8_t>(score::lcm::ProcessState::kTerminating),
off = static_cast<uint8_t>(score::lcm::ProcessState::kTerminated)};
off = static_cast<uint8_t>(score::lcm::ProcessState::kTerminated),
failed = static_cast<uint8_t>(score::lcm::ProcessState::kFailed)};
Comment thread
WilliamRoebuck marked this conversation as resolved.

/// @brief Get Process State
/// @return Returns Process State
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,9 @@ constexpr ProcessState::EProcState ProcessStateReader::translateProcessState(
static_assert(static_cast<uint8_t>(ProcessState::EProcState::off) ==
static_cast<uint8_t>(score::lcm::ProcessState::kTerminated),
"Lcm State Enum and ProcessState::EProcState Enum do not match.");
static_assert(static_cast<uint8_t>(ProcessState::EProcState::failed) ==
static_cast<uint8_t>(score::lcm::ProcessState::kFailed),
"Lcm State Enum and ProcessState::EProcState Enum do not match.");
return static_cast<ProcessState::EProcState>(f_processStateLcm);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ void Alive::updateData(const ifexm::ProcessState& f_observable_r) noexcept(true)

const bool isRelevant = (state == ifexm::ProcessState::EProcState::running) ||
(state == ifexm::ProcessState::EProcState::sigterm) ||
(state == ifexm::ProcessState::EProcState::failed) ||
(state == ifexm::ProcessState::EProcState::off);

if (isRelevant)
Expand Down Expand Up @@ -279,7 +280,8 @@ Alive::EUpdateEventType Alive::getAliveEventType(bool f_isEvaluationEvent,
return EUpdateEventType::kActivation;
}
if (snapshot.eProcState == ifexm::ProcessState::EProcState::sigterm ||
snapshot.eProcState == ifexm::ProcessState::EProcState::off)
snapshot.eProcState == ifexm::ProcessState::EProcState::off ||
snapshot.eProcState == ifexm::ProcessState::EProcState::failed)
{
return EUpdateEventType::kDeactivation;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,13 @@
#include <memory>
#include <optional>

#include "score/mw/launch_manager/common/identifier_hash.hpp"
#include "score/mw/launch_manager/recovery_client/irecovery_client.h"
#include "score/mw/launch_manager/alive_monitor/details/ifappl/Checkpoint.hpp"
#include "score/mw/launch_manager/alive_monitor/details/ifexm/ProcessCfg.hpp"
#include "score/mw/launch_manager/alive_monitor/details/ifexm/ProcessState.hpp"
#include "score/mw/launch_manager/alive_monitor/details/supervision/Alive.hpp"
#include "score/mw/launch_manager/alive_monitor/details/supervision/SupervisionCfg.hpp"
#include "score/mw/launch_manager/common/identifier_hash.hpp"
#include "score/mw/launch_manager/recovery_client/irecovery_client.h"

using namespace testing;

Expand All @@ -36,12 +36,16 @@ namespace
class MockRecoveryClient : public score::lcm::IRecoveryClient
{
public:
MOCK_METHOD(bool,
sendRecoveryRequest,
(const score::lcm::IdentifierHash& process_group_identifier),
(noexcept, override));
MOCK_METHOD(std::optional<score::lcm::IdentifierHash>, getNextRequest, (), (noexcept, override));
MOCK_METHOD(bool, hasOverflow, (), (const, noexcept, override));
MOCK_METHOD(
void,
setRecoveryRequestCallback,
(score::lcm::IRecoveryClient::RecoveryRequestCallback callback),
(noexcept, override));
MOCK_METHOD(
bool,
sendRecoveryRequest,
(const score::lcm::IdentifierHash& process_group_identifier),
(noexcept, override));
};

/// Helper: build a minimal Alive under test.
Expand Down Expand Up @@ -168,10 +172,11 @@ class AliveSupervisionTest : public ::testing::Test

TEST_F(AliveSupervisionTest, AliveTransitionsOkToExpiredOnMissingHeartbeat)
{
RecordProperty("Description",
"Verify that Alive transitions from deactivated -> ok -> expired when no heartbeats are reported "
"and failedCyclesTolerance == 0. sendRecoveryRequest must be called exactly once with the "
"configured recovery target hash.");
RecordProperty(
"Description",
"Verify that Alive transitions from deactivated -> ok -> expired when no heartbeats are reported "
"and failedCyclesTolerance == 0. sendRecoveryRequest must be called exactly once with the "
"configured recovery target hash.");
AliveFixture fix = AliveFixture::Builder{}.build();

EXPECT_CALL(*fix.mockClient, sendRecoveryRequest(fix.kProcessIdentifier)).Times(1).WillOnce(Return(true));
Expand All @@ -189,8 +194,8 @@ TEST_F(AliveSupervisionTest, AliveTransitionsOkToExpiredOnMissingHeartbeat)

TEST_F(AliveSupervisionTest, AliveStaysOkWithCorrectHeartbeats)
{
RecordProperty("Description",
"Verify that sending at least minIndications heartbeats per cycle keeps Alive in ok.");
RecordProperty(
"Description", "Verify that sending at least minIndications heartbeats per cycle keeps Alive in ok.");
AliveFixture fix = AliveFixture::Builder{}.build();

EXPECT_CALL(*fix.mockClient, sendRecoveryRequest(_)).Times(0);
Expand Down Expand Up @@ -232,9 +237,10 @@ TEST_F(AliveSupervisionTest, AliveReportsEnqueueFailureWhenRingBufferFull)

TEST_F(AliveSupervisionTest, AliveDebouncesThroughFailedBeforeExpired)
{
RecordProperty("Description",
"Verify that failedCyclesTolerance debouncing works: with tolerance=1 the supervision passes "
"through failed before reaching expired.");
RecordProperty(
"Description",
"Verify that failedCyclesTolerance debouncing works: with tolerance=1 the supervision passes "
"through failed before reaching expired.");
AliveFixture fix = AliveFixture::Builder{}.withFailedCyclesTolerance(1U).build();

EXPECT_CALL(*fix.mockClient, sendRecoveryRequest(fix.kProcessIdentifier))
Expand All @@ -256,8 +262,7 @@ TEST_F(AliveSupervisionTest, AliveDebouncesThroughFailedBeforeExpired)

TEST_F(AliveSupervisionTest, DeactivatesOnProcessSigterm)
{
RecordProperty("Description",
"Verify that a clean shutdown (sigterm) deactivates the supervision from ok.");
RecordProperty("Description", "Verify that a clean shutdown (sigterm) deactivates the supervision from ok.");
AliveFixture fix = AliveFixture::Builder{}.build();

EXPECT_CALL(*fix.mockClient, sendRecoveryRequest(_)).Times(0);
Expand All @@ -273,8 +278,8 @@ TEST_F(AliveSupervisionTest, DeactivatesOnProcessSigterm)

TEST_F(AliveSupervisionTest, DeactivatesOnProcessCrash)
{
RecordProperty("Description",
"Verify that a process crash (off without sigterm) also deactivates the supervision.");
RecordProperty(
"Description", "Verify that a process crash (off without sigterm) also deactivates the supervision.");
AliveFixture fix = AliveFixture::Builder{}.build();

EXPECT_CALL(*fix.mockClient, sendRecoveryRequest(_)).Times(0);
Expand All @@ -290,9 +295,10 @@ TEST_F(AliveSupervisionTest, DeactivatesOnProcessCrash)

TEST_F(AliveSupervisionTest, ReactivatesAfterCrash)
{
RecordProperty("Description",
"Verify that after a crash (off) the supervision can be reactivated when the process"
" reports running again, without any special recovery path.");
RecordProperty(
"Description",
"Verify that after a crash (off) the supervision can be reactivated when the process"
" reports running again, without any special recovery path.");
AliveFixture fix = AliveFixture::Builder{}.build();

EXPECT_CALL(*fix.mockClient, sendRecoveryRequest(_)).Times(0);
Expand All @@ -314,9 +320,10 @@ TEST_F(AliveSupervisionTest, ReactivatesAfterCrash)

TEST_F(AliveSupervisionTest, IgnoresIrrelevantProcessStates)
{
RecordProperty("Description",
"Verify that process states other than running/sigterm/off are ignored and do not"
" affect the supervision state.");
RecordProperty(
"Description",
"Verify that process states other than running/sigterm/off are ignored and do not"
" affect the supervision state.");
AliveFixture fix = AliveFixture::Builder{}.build();

EXPECT_CALL(*fix.mockClient, sendRecoveryRequest(_)).Times(0);
Expand All @@ -341,14 +348,11 @@ TEST_F(AliveSupervisionTest, IgnoresIrrelevantProcessStates)

TEST_F(AliveSupervisionTest, MaxIndicationViolationExpires)
{
RecordProperty("Description",
"Verify that exceeding the maximum allowed heartbeats per cycle leads to failure.");
RecordProperty("Description", "Verify that exceeding the maximum allowed heartbeats per cycle leads to failure.");
// max=1, tolerance=0: more than 1 heartbeat per cycle expires immediately
AliveFixture fix = AliveFixture::Builder{}.withMaxIndications(1U).build();

EXPECT_CALL(*fix.mockClient, sendRecoveryRequest(fix.kProcessIdentifier))
.Times(1)
.WillOnce(Return(true));
EXPECT_CALL(*fix.mockClient, sendRecoveryRequest(fix.kProcessIdentifier)).Times(1).WillOnce(Return(true));

fix.activateProcess(10U);
fix.alive->evaluate(11U);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ cc_library(
":mpmc_concurrent_queue",
"//score/launch_manager/src/daemon/src/common:constants",
"//score/launch_manager/src/daemon/src/common:log",
"//score/launch_manager/src/daemon/src/process_group_manager/details:icomponent_controller",
],
)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,28 +17,33 @@
#include "score/mw/launch_manager/common/concurrency/mpmc_concurrent_queue.hpp"
#include "score/mw/launch_manager/common/constants.hpp"
#include "score/mw/launch_manager/common/log.hpp"
#include "score/mw/launch_manager/process_group_manager/details/icomponent_controller.hpp"
#include <memory>
#include <thread>
#include <vector>

namespace score::lcm::internal
{

using namespace score::mw::lifecycle::internal;

/// @brief Templated worker thread pool for executing jobs from a queue.
/// This class manages a pool of worker threads that continuously retrieve and execute jobs
/// from an MPMCConcurrentQueue until the pool is stopped or destructed.
/// @tparam T The type of items stored in the queue (as std::shared_ptr<T>).
template <class T>
class WorkerThread final
{
using Queue = MPMCConcurrentQueue<std::shared_ptr<T>, static_cast<std::size_t>(ProcessLimits::kMaxProcesses)>;
using Queue = MPMCConcurrentQueue<std::optional<T>, static_cast<std::size_t>(ProcessLimits::kMaxProcesses)>;

public:
/// @brief Constructs a WorkerThread pool with the specified number of threads.
///
/// @param queue The MpmcQueue from which threads will take work items.
/// @param num_threads Number of threads in the pool.
WorkerThread(std::shared_ptr<Queue> queue, uint32_t num_threads) : the_job_queue_(queue)
/// @param component_controller_ The controller to delegate work to.
WorkerThread(std::shared_ptr<Queue> queue, uint32_t num_threads, IComponentController& component_controller)
: the_job_queue_(queue), component_controller_(component_controller)
{
worker_threads_.reserve(num_threads);
for (uint32_t i = 0U; i < num_threads; ++i)
Expand Down Expand Up @@ -99,13 +104,15 @@ class WorkerThread final
LM_LOG_ERROR() << "Got an error getting a job: " << job.error();
continue;
}
(*job)->doWork();
component_controller_.doWork(std::move(**job));
}
}

/// @brief The queue from which each thread takes work.
std::shared_ptr<Queue> the_job_queue_{};

IComponentController& component_controller_;

/// @brief Vector of worker threads.
std::vector<std::unique_ptr<std::thread>> worker_threads_{};
};
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -272,7 +272,6 @@ ProcessGroupState ConfigurationAdapter::buildProcessGroupState(
std::vector<ProcessGroupState> ConfigurationAdapter::buildProcessGroupStates(
const Config& config) const {
std::vector<ProcessGroupState> states;
states.push_back(ProcessGroupState{IdentifierHash{"MainPG/Off"}, {}});

const auto& run_targets = config.runTargets();
DependsOnMap depends_on_by_name;
Expand All @@ -288,6 +287,17 @@ std::vector<ProcessGroupState> ConfigurationAdapter::buildProcessGroupStates(
}

states.push_back(buildProcessGroupState("MainPG/fallback_run_target", fallback_cfg.depends_on, depends_on_by_name));

// Guarantee an Off state exists so it always has a RunTarget node. Add an empty one (no
// processes == everything stopped) unless a run target already produced it.
const IdentifierHash off_name{"MainPG/Off"}; // matches pg.off_state_ set in buildFromConfig
const bool has_off = std::any_of(
states.cbegin(), states.cend(), [&](const ProcessGroupState& s) { return s.name_ == off_name; });
if (!has_off)
{
states.push_back(ProcessGroupState{off_name, {}});
}

return states;
}

Expand Down Expand Up @@ -405,6 +415,19 @@ std::optional<const std::vector<uint32_t>*> ConfigurationAdapter::getProcessInde
return std::nullopt;
}

std::optional<const std::vector<ProcessGroupState>*> ConfigurationAdapter::getListOfProcessGroupStates(
const IdentifierHash& pg_name) const
{
std::optional<const std::vector<ProcessGroupState>*> result = std::nullopt;

if (const auto* pg = getProcessGroupByID(pg_name))
{
result = &pg->states_;
}

return result;
}

std::optional<const OsProcess*> ConfigurationAdapter::getOsProcessConfiguration(
const IdentifierHash& pg_name, const uint32_t index) const {
if (auto pg = getProcessGroupByNameAndIndex(pg_name, index)) {
Expand Down
Loading
Loading