From a719452d049f373c5dc3dada904c045c93837bab Mon Sep 17 00:00:00 2001 From: Mykola Kobets Date: Tue, 10 Feb 2026 19:05:22 +0200 Subject: [PATCH 1/4] cm: launcher: rebalance on node info change Signed-off-by: Mykola Kobets Reviewed-by: Oleksandr Grytsov Reviewed-by: Mykola Solianko --- src/core/cm/launcher/launcher.cpp | 15 +++++++++------ src/core/cm/launcher/launcher.hpp | 1 + 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/src/core/cm/launcher/launcher.cpp b/src/core/cm/launcher/launcher.cpp index 2e939082f..e8805d846 100644 --- a/src/core/cm/launcher/launcher.cpp +++ b/src/core/cm/launcher/launcher.cpp @@ -374,7 +374,8 @@ void Launcher::ProcessUpdate() UniqueLock updateLock {mUpdateMutex}; mProcessUpdatesCondVar.Wait(updateLock, [this]() { - return (!mUpdatedNodes.IsEmpty() || mNewSubjects.HasValue() || mAlertReceived || !mIsRunning) + return (!mUpdatedNodes.IsEmpty() || mNewSubjects.HasValue() || mAlertReceived || !mIsRunning + || mIsNodeInfoChanged) && !mDisableProcessUpdates; }); @@ -407,6 +408,12 @@ void Launcher::ProcessUpdate() doRebalance = true; } + // Process node info changed. + if (mIsNodeInfoChanged) { + mIsNodeInfoChanged = false; + doRebalance = true; + } + // Resend instances. if (!mUpdatedNodes.IsEmpty()) { if (!doRebalance) { @@ -593,11 +600,7 @@ void Launcher::OnNodeInfoChanged(const UnitNodeInfo& info) UniqueLock updateLock {mUpdateMutex}; if (mNodeManager.UpdateNodeInfo(info)) { - if (auto err = PushUnique(mUpdatedNodes, info.mNodeID); !err.IsNone()) { - LOG_ERR() << "Failed to add node ID to updated nodes" << Log::Field(AOS_ERROR_WRAP(err)); - - return; - } + mIsNodeInfoChanged = true; mProcessUpdatesCondVar.NotifyAll(); mAllNodesConnectedCondVar.NotifyAll(); diff --git a/src/core/cm/launcher/launcher.hpp b/src/core/cm/launcher/launcher.hpp index 6723e8885..b7c8a3612 100644 --- a/src/core/cm/launcher/launcher.hpp +++ b/src/core/cm/launcher/launcher.hpp @@ -188,6 +188,7 @@ class Launcher : public LauncherItf, bool mDisableProcessUpdates {}; StaticArray, cMaxNumNodes> mUpdatedNodes; bool mAlertReceived {}; + bool mIsNodeInfoChanged {}; Optional mNewSubjects; // Misc From 7d23f8cbbb9b6cd58099f5f0e764b686c4cbffea Mon Sep 17 00:00:00 2001 From: Mykola Kobets Date: Tue, 10 Feb 2026 18:26:16 +0200 Subject: [PATCH 2/4] cm: launcher: send diff for update statuses Signed-off-by: Mykola Kobets Reviewed-by: Oleksandr Grytsov Reviewed-by: Mykola Solianko --- src/core/cm/launcher/launcher.cpp | 27 +++++++++++++++++++++++-- src/core/cm/launcher/launcher.hpp | 2 +- src/core/cm/launcher/tests/launcher.cpp | 3 +-- 3 files changed, 27 insertions(+), 5 deletions(-) diff --git a/src/core/cm/launcher/launcher.cpp b/src/core/cm/launcher/launcher.cpp index e8805d846..f33cce591 100644 --- a/src/core/cm/launcher/launcher.cpp +++ b/src/core/cm/launcher/launcher.cpp @@ -285,6 +285,15 @@ void Launcher::UpdateInstanceStatuses() const auto& preinstalledComponents = mInstanceManager.GetPreinstalledComponents(); const auto totalSize = activeInstances.Size() + preinstalledComponents.Size(); + // Copy old statuses. + auto oldInstanceStatuses = MakeUnique>(&mAllocator); + if (auto err = oldInstanceStatuses->Assign(mInstanceStatuses); !err.IsNone()) { + LOG_ERR() << "Failed to copy old instance statuses" << Log::Field(AOS_ERROR_WRAP(err)); + + return; + } + + // Keep current statuses. bool changed = mInstanceStatuses.Size() != totalSize; if (auto err = mInstanceStatuses.Resize(totalSize); !err.IsNone()) { @@ -316,7 +325,21 @@ void Launcher::UpdateInstanceStatuses() return; } - for (const auto& status : mInstanceStatuses) { + // Find new statuses. + auto changedStatuses = MakeUnique>(&mAllocator); + + for (size_t i = 0; i < mInstanceStatuses.Size(); ++i) { + auto newStatus = !oldInstanceStatuses->Contains(mInstanceStatuses[i]); + if (newStatus) { + if (auto err = changedStatuses->PushBack(mInstanceStatuses[i]); !err.IsNone()) { + LOG_ERR() << "Failed to add changed status" << Log::Field(AOS_ERROR_WRAP(err)); + + return; + } + } + } + + for (const auto& status : *changedStatuses) { LOG_INF() << "Instance status changed" << Log::Field("instance", static_cast(status)) << Log::Field("version", status.mVersion) << Log::Field("nodeID", status.mNodeID) << Log::Field("runtimeID", status.mRuntimeID) << Log::Field("manifestDigest", status.mManifestDigest) @@ -324,7 +347,7 @@ void Launcher::UpdateInstanceStatuses() } for (auto& listener : mInstanceStatusListeners) { - listener->OnInstancesStatusesChanged(mInstanceStatuses); + listener->OnInstancesStatusesChanged(*changedStatuses); } } diff --git a/src/core/cm/launcher/launcher.hpp b/src/core/cm/launcher/launcher.hpp index b7c8a3612..3c240b39c 100644 --- a/src/core/cm/launcher/launcher.hpp +++ b/src/core/cm/launcher/launcher.hpp @@ -134,7 +134,7 @@ class Launcher : public LauncherItf, private: static constexpr auto cMaxNumInstanceStatusListeners = 8; - static constexpr auto cAllocatorSize = sizeof(StaticArray) + static constexpr auto cAllocatorSize = 2 * sizeof(StaticArray) + sizeof(StaticArray, cMaxNumInstances>); void SendRunStatus(); diff --git a/src/core/cm/launcher/tests/launcher.cpp b/src/core/cm/launcher/tests/launcher.cpp index e90d0ecbd..d2ad717b6 100644 --- a/src/core/cm/launcher/tests/launcher.cpp +++ b/src/core/cm/launcher/tests/launcher.cpp @@ -1454,7 +1454,6 @@ TEST_F(CMLauncherTest, Balancing) ASSERT_TRUE(mLauncher.RunInstances(testItem.mRunRequests, *runStatuses).IsNone()); ASSERT_EQ(*runStatuses, testItem.mExpectedRunStatus); - ASSERT_EQ(instanceStatusListener.GetLastStatuses(), testItem.mExpectedRunStatus); // Rebalance if (testItem.mRebalancing) { @@ -1584,7 +1583,7 @@ TEST_F(CMLauncherTest, PlatformFiltering) expectedRunStatus->PushBack(CreateInstanceStatus(CreateInstanceIdent(cService3, cSubject1, 0), cNodeIDRemoteSM2, cRunnerRunc, aos::InstanceStateEnum::eActive, ErrorEnum::eNone, "", false, manifestDigest.CStr())); - ASSERT_EQ(instanceStatusListener.GetLastStatuses(), *expectedRunStatus); + ASSERT_EQ(*runStatuses, *expectedRunStatus); } TEST_F(CMLauncherTest, ResendInstancesOnMismatchedNodeStatus) From f23eeb2c32c25081b79bed0bd575b30ae73f1277 Mon Sep 17 00:00:00 2001 From: Mykola Kobets Date: Wed, 11 Feb 2026 11:26:58 +0200 Subject: [PATCH 3/4] cm: launcher: skip inactive instances in process updates Signed-off-by: Mykola Kobets Reviewed-by: Oleksandr Grytsov Reviewed-by: Mykola Solianko --- src/core/cm/launcher/instancemanager.cpp | 7 ++++++- src/core/cm/launcher/node.cpp | 10 ++++++---- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/src/core/cm/launcher/instancemanager.cpp b/src/core/cm/launcher/instancemanager.cpp index 3ead20f0c..a03d6a6e5 100644 --- a/src/core/cm/launcher/instancemanager.cpp +++ b/src/core/cm/launcher/instancemanager.cpp @@ -161,7 +161,12 @@ Error InstanceManager::UpdateStatus(const InstanceStatus& status) auto instance = FindActiveInstance(status); if (!instance) { - // not expected instance received from SM. + // Ignore inactive instance, SM sometimes sends inactive status for stopped instances. + if (status.mState == aos::InstanceStateEnum::eInactive) { + return ErrorEnum::eNone; + } + + // Not expected instance received from SM. return AOS_ERROR_WRAP(ErrorEnum::eNotFound); } diff --git a/src/core/cm/launcher/node.cpp b/src/core/cm/launcher/node.cpp index 4966df835..24faeb1ec 100644 --- a/src/core/cm/launcher/node.cpp +++ b/src/core/cm/launcher/node.cpp @@ -75,9 +75,11 @@ class Filter { Cmp mCmp; }; -auto FilterByNode(const Array& array, const String& nodeID) +auto FilterActiveNodeInstances(const Array& array, const String& nodeID) { - auto cmp = [nodeID](const InstanceStatus& status) { return status.mNodeID == nodeID; }; + auto cmp = [nodeID](const InstanceStatus& status) { + return status.mNodeID == nodeID && status.mState != aos::InstanceStateEnum::eInactive; + }; return Filter(array, cmp); } @@ -309,7 +311,7 @@ Error Node::SendScheduledInstances( auto stopInstances = MakeUnique>(mAllocator); auto startInstances = MakeUnique>(mAllocator); - for (const auto& status : FilterByNode(runningInstances, mInfo.mNodeID)) { + for (const auto& status : FilterActiveNodeInstances(runningInstances, mInfo.mNodeID)) { // Check if the instance is scheduled on this node. auto isScheduled = scheduledInstances.ContainsIf([&status, this](const SharedPtr& item) { return static_cast(status) == item->GetInfo().mInstanceIdent @@ -359,7 +361,7 @@ RetWithError Node::ResendInstances( auto startInstances = MakeUnique>(mAllocator); size_t runningNodeInstances = 0; - for (const auto& status : FilterByNode(runningInstances, mInfo.mNodeID)) { + for (const auto& status : FilterActiveNodeInstances(runningInstances, mInfo.mNodeID)) { runningNodeInstances++; auto isActive = activeInstances.ContainsIf([&status](const SharedPtr& item) { From 93e2de840d339ed882788500bfe547c625b076a0 Mon Sep 17 00:00:00 2001 From: Mykola Kobets Date: Wed, 11 Feb 2026 11:27:40 +0200 Subject: [PATCH 4/4] cm: launcher: fix node instance status log Signed-off-by: Mykola Kobets Reviewed-by: Oleksandr Grytsov Reviewed-by: Mykola Solianko --- src/core/cm/launcher/balancer.cpp | 4 ++-- src/core/cm/launcher/launcher.cpp | 9 ++++++--- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/src/core/cm/launcher/balancer.cpp b/src/core/cm/launcher/balancer.cpp index 7efd44fa2..c456cb7f1 100644 --- a/src/core/cm/launcher/balancer.cpp +++ b/src/core/cm/launcher/balancer.cpp @@ -46,8 +46,8 @@ Error Balancer::RunInstances(UniqueLock& lock, Array> return AOS_ERROR_WRAP(err); } - // Submit scheduled instances before sending instances to nodes. - // So following status updates will change active instances. + // Submit scheduled instances before sending them to nodes. + // So status updates expecting by SendScheduledInstances will be assigned to active instances. if (auto err = mInstanceManager->SubmitScheduledInstances(); !err.IsNone()) { return AOS_ERROR_WRAP(err); } diff --git a/src/core/cm/launcher/launcher.cpp b/src/core/cm/launcher/launcher.cpp index f33cce591..06b90ff87 100644 --- a/src/core/cm/launcher/launcher.cpp +++ b/src/core/cm/launcher/launcher.cpp @@ -575,12 +575,15 @@ Error Launcher::OnInstanceStatusReceived(const InstanceStatus& status) Error Launcher::OnNodeInstancesStatusesReceived(const String& nodeID, const Array& statuses) { + LOG_INF() << "Node instances statuses received" << Log::Field("nodeID", nodeID) + << Log::Field("numStatuses", statuses.Size()); + for (const auto& status : statuses) { LOG_INF() << "Node instance status received" << Log::Field("instance", static_cast(status)) - << Log::Field("version", status.mVersion) << Log::Field("nodeID", status.mNodeID) - << Log::Field("runtimeID", status.mRuntimeID) << Log::Field("manifestDigest", status.mManifestDigest) - << Log::Field("state", status.mState) << Log::Field(status.mError); + << Log::Field("version", status.mVersion) << Log::Field("runtimeID", status.mRuntimeID) + << Log::Field("manifestDigest", status.mManifestDigest) << Log::Field("state", status.mState) + << Log::Field(status.mError); } LockGuard updateLock {mUpdateMutex};