Network batching

src/core/sm/launcher/launcher.cpp

@@ -792,6 +792,11 @@
 {
     LOG_INF() << "[profiling] Stop all networks begin" << Log::Field("count", mInstances.Size());
 
+    auto errBegin = mNetworkManager->BeginBatch();
+    if (!errBegin.IsNone()) {
+        LOG_ERR() << "Failed to begin network batch" << Log::Field(AOS_ERROR_WRAP(errBegin));
+    }
+
     for (auto& instance : mInstances) {
         if (instance.mInfo.mType != UpdateItemTypeEnum::eService) {
             continue;
@@ -804,6 +809,20 @@
         }
     }
 
+    if (auto err = mLaunchPool.Wait(); !err.IsNone()) {
+        LOG_ERR() << "Thread pool wait failed" << Log::Field(AOS_ERROR_WRAP(err));
+    }
+
+    if (errBegin.IsNone()) {
+        auto failedIDs = MakeUnique<StaticArray<StaticString<cIDLen>, cMaxNumInstances>>(&mAllocator);
+
+        mNetworkManager->FlushBatch(*failedIDs);
+
+        if (!failedIDs->IsEmpty()) {
+            LOG_WRN() << "Network stop batch partially failed" << Log::Field("count", failedIDs->Size());
+        }
+    }
+
     LOG_INF() << "[profiling] Stop all networks end";
 }
 
@@ -900,6 +919,11 @@
 {
     LOG_INF() << "[profiling] Start networks begin" << Log::Field("count", startInstances.Size());
 
+    auto errBegin = mNetworkManager->BeginBatch();
+    if (!errBegin.IsNone()) {
+        LOG_ERR() << "Failed to begin network batch" << Log::Field(AOS_ERROR_WRAP(errBegin));
+    }
+
     for (const auto& instance : startInstances) {
         auto instanceData = FindInstanceData(instance);
         if (!instanceData) {
@@ -932,6 +956,29 @@
         LOG_ERR() << "Thread pool wait failed" << Log::Field(AOS_ERROR_WRAP(err));
     }
 
+    if (errBegin.IsNone()) {
+        auto failedIDs = MakeUnique<StaticArray<StaticString<cIDLen>, cMaxNumInstances>>(&mAllocator);
+
+        mNetworkManager->FlushBatch(*failedIDs);
+
+        for (const auto& failedID : *failedIDs) {
+            auto instanceData = FindInstanceDataByID(failedID);
+            if (!instanceData) {
+                continue;
+            }
+
+            SetInstanceState(*instanceData, InstanceStateEnum::eFailed,
+                AOS_ERROR_WRAP(Error(ErrorEnum::eFailed, "network batch apply failed")));
+
+            if (auto err
+                = mNetworkManager->StopInstanceNetwork(instanceData->mInstanceID, instanceData->mInfo.mOwnerID);
+                !err.IsNone() && !err.Is(ErrorEnum::eNotFound)) {
+                LOG_ERR() << "Failed to stop network" << Log::Field("instance", instanceData->mInfo)
+                          << Log::Field(AOS_ERROR_WRAP(err));
+            }
+        }
+    }
+
     LOG_INF() << "[profiling] Start networks end";
 }
 
@@ -957,6 +1004,11 @@
 {
     LOG_INF() << "[profiling] Stop networks begin" << Log::Field("count", stopInstances.Size());
 
+    auto errBegin = mNetworkManager->BeginBatch();
+    if (!errBegin.IsNone()) {
+        LOG_ERR() << "Failed to begin network batch" << Log::Field(AOS_ERROR_WRAP(errBegin));
+    }
+
     for (const auto& instance : stopInstances) {
         auto instanceData = FindInstanceData(instance);
         if (!instanceData) {
@@ -981,6 +1033,16 @@
         LOG_ERR() << "Thread pool wait failed" << Log::Field(AOS_ERROR_WRAP(err));
     }
 
+    if (errBegin.IsNone()) {
+        auto failedIDs = MakeUnique<StaticArray<StaticString<cIDLen>, cMaxNumInstances>>(&mAllocator);
+
+        mNetworkManager->FlushBatch(*failedIDs);
+
+        if (!failedIDs->IsEmpty()) {
+            LOG_WRN() << "Network stop batch partially failed" << Log::Field("count", failedIDs->Size());
+        }
+    }
+
     LOG_INF() << "[profiling] Stop networks end";
 }
 
@@ -1127,6 +1189,16 @@
     return const_cast<Launcher*>(this)->FindInstanceData(instanceIdent);
 }
 
+Launcher::InstanceData* Launcher::FindInstanceDataByID(const String& instanceID)
+{
+    auto it = mInstances.FindIf([&instanceID](const auto& instance) { return instance.mInstanceID == instanceID; });
+    if (it != mInstances.end()) {
+        return it;
+    }
+
+    return nullptr;
+}
+
 RuntimeItf* Launcher::FindInstanceRuntime(const String& runtimeID)
 {
     auto it = mRuntimes.FindIf([&runtimeID](const auto& it) { return it.mSecond == runtimeID; });

src/core/sm/launcher/launcher.hpp

@@ -225,6 +225,7 @@ class Launcher : public LauncherItf,
 
     InstanceData*      FindInstanceData(const InstanceIdent& instanceIdent);
     InstanceData*      FindInstanceData(const InstanceIdent& instanceIdent) const;
+    InstanceData*      FindInstanceDataByID(const String& instanceID);
     RuntimeItf*        FindInstanceRuntime(const String& runtimeID);
     RuntimeItf*        FindInstanceRuntime(const String& runtimeID) const;
     RuntimeItf*        FindInstanceRuntime(const InstanceIdent& instanceIdent);

src/core/sm/launcher/tests/launcher.cpp

@@ -194,6 +194,8 @@ class LauncherTest : public Test {
         EXPECT_CALL(mNetworkManager, StartInstanceNetwork).WillRepeatedly(Return(ErrorEnum::eNone));
         EXPECT_CALL(mNetworkManager, StopInstanceNetwork).WillRepeatedly(Return(ErrorEnum::eNone));
         EXPECT_CALL(mNetworkManager, ReleaseInstanceNetwork).WillRepeatedly(Return(ErrorEnum::eNone));
+        EXPECT_CALL(mNetworkManager, BeginBatch()).WillRepeatedly(Return(ErrorEnum::eNone));
+        EXPECT_CALL(mNetworkManager, FlushBatch(_)).WillRepeatedly(Return(ErrorEnum::eNone));
     }
 
     StaticArray<RuntimeItf*, cMaxNumNodeRuntimes> GetRuntimesArray()
@@ -462,6 +464,65 @@ TEST_F(LauncherTest, LauncherStartsStoredInstancesOnModuleStart)
     ASSERT_TRUE(err.IsNone()) << tests::utils::ErrorToStr(err);
 }
 
+TEST_F(LauncherTest, StartNetworks_FlushFailure_FailsInstance)
+{
+    const std::vector cStoredInfos = {
+        CreateInstanceInfo("item0", 0, "1.0.0", "runtime0"),
+        CreateInstanceInfo("item1", 1, "1.0.0", "runtime1"),
+    };
+
+    mStorage.Init(cStoredInfos);
+
+    auto err = mLauncher.Init(GetRuntimesArray(), mImageManager, mSender, mStorage, mOCISpec, mItemInfoProvider,
+        mCloudConnection, mNetworkManager, mInstanceIDProvider, mResourceInfoProvider);
+    ASSERT_TRUE(err.IsNone()) << tests::utils::ErrorToStr(err);
+
+    EXPECT_CALL(mInstanceIDProvider, GetInstanceID)
+        .WillRepeatedly(Invoke([](const InstanceIdent& instance, String& instanceID) {
+            instanceID = instance.mItemID;
+
+            return ErrorEnum::eNone;
+        }));
+
+    EXPECT_CALL(mNetworkManager, FlushBatch(_))
+        .WillRepeatedly(DoAll(WithArg<0>([](auto& failedInstanceIDs) { failedInstanceIDs.PushBack("item0"); }),
+            Return(ErrorEnum::eNone)));
+
+    EXPECT_CALL(mNetworkManager, StopInstanceNetwork(String("item0"), _))
+        .Times(AtLeast(1))
+        .WillRepeatedly(Return(ErrorEnum::eNone));
+
+    EXPECT_CALL(mRuntime1, StartInstance).WillOnce(Invoke([](const InstanceInfo& instance, InstanceStatus& status) {
+        SetInstanceStatus(instance, InstanceStateEnum::eActive, status);
+
+        return ErrorEnum::eNone;
+    }));
+
+    err = mLauncher.Start();
+    ASSERT_TRUE(err.IsNone()) << tests::utils::ErrorToStr(err);
+
+    err = mLauncher.GetInstancesStatuses(mReceivedStatuses);
+    ASSERT_TRUE(err.IsNone()) << tests::utils::ErrorToStr(err);
+
+    ASSERT_EQ(mReceivedStatuses.Size(), cStoredInfos.size());
+
+    for (const auto& status : mReceivedStatuses) {
+        if (status.mItemID == "item0") {
+            EXPECT_EQ(status.mState, InstanceStateEnum::eFailed);
+        } else {
+            EXPECT_EQ(status.mState, InstanceStateEnum::eActive);
+        }
+    }
+
+    EXPECT_CALL(mRuntime0, StopInstance(static_cast<const InstanceIdent&>(cStoredInfos[0]), _))
+        .WillOnce(Return(ErrorEnum::eNone));
+    EXPECT_CALL(mRuntime1, StopInstance(static_cast<const InstanceIdent&>(cStoredInfos[1]), _))
+        .WillOnce(Return(ErrorEnum::eNone));
+
+    err = mLauncher.Stop();
+    ASSERT_TRUE(err.IsNone()) << tests::utils::ErrorToStr(err);
+}
+
 TEST_F(LauncherTest, StopInstancesWithExpiredOfflineTTL)
 {
     const std::vector cStoredInfos = {

src/core/sm/networkmanager/itf/firewall.hpp

@@ -158,6 +158,35 @@ class FirewallItf {
      * @return Error.
      */
     virtual Error RemoveMasquerade(const String& subnet, const String& outIf) = 0;
+
+    /**
+     * Opens a batch; AddInstance/RemoveInstance calls are staged until flush.
+     *
+     * @return Error.
+     */
+    virtual Error BeginBatch() = 0;
+
+    /**
+     * Flushes the staged batch atomically in a single nft transaction.
+     *
+     * @return Error.
+     */
+    virtual Error FlushBatch() = 0;
+
+    /**
+     * Discards the staged batch and leaves batch mode without applying anything to the kernel
+     * (unlike FlushBatch which commits, or Revert which undoes an already-applied batch).
+     *
+     * @return Error.
+     */
+    virtual Error AbortBatch() = 0;
+
+    /**
+     * Reverts the flushed batch, deleting everything it applied by handle.
+     *
+     * @return Error.
+     */
+    virtual Error Revert() = 0;
 };
 
 /** @}*/

src/core/sm/networkmanager/itf/networkmanager.hpp

@@ -128,6 +128,21 @@ class NetworkManagerItf : public SystemTrafficProviderItf,
      * @return Error.
      */
     virtual Error ReleaseInstanceNetwork(const String& instanceID, const String& networkID) = 0;
+
+    /**
+     * Opens a batch; start/stop operations are staged and applied on flush.
+     *
+     * @return Error.
+     */
+    virtual Error BeginBatch() = 0;
+
+    /**
+     * Flushes the staged batch atomically across firewall, traffic and storage.
+     *
+     * @param[out] failedInstanceIDs instances that were not applied.
+     * @return Error.
+     */
+    virtual Error FlushBatch(Array<StaticString<cIDLen>>& failedInstanceIDs) = 0;
 };
 
 /** @}*/

src/core/sm/networkmanager/itf/storage.hpp

@@ -202,6 +202,27 @@ class StorageItf {
      */
     virtual Error RemoveTrafficMonitorData(const String& chain) = 0;
 
+    /**
+     * Begins a storage transaction; subsequent writes are staged until commit.
+     *
+     * @return Error.
+     */
+    virtual Error BeginTransaction() = 0;
+
+    /**
+     * Commits the current storage transaction.
+     *
+     * @return Error.
+     */
+    virtual Error CommitTransaction() = 0;
+
+    /**
+     * Rolls back the current storage transaction, discarding staged writes.
+     *
+     * @return Error.
+     */
+    virtual Error RollbackTransaction() = 0;
+
     /**
      * Destroys storage interface.
      */

src/core/sm/networkmanager/itf/trafficmonitor.hpp

@@ -104,6 +104,35 @@ class TrafficMonitorItf {
      */
     virtual Error GetInstanceTraffic(const String& instanceID, uint64_t& inputTraffic, uint64_t& outputTraffic) const
         = 0;
+
+    /**
+     * Opens a batch; StartInstanceMonitoring/StopInstanceMonitoring calls are staged until flush.
+     *
+     * @return Error.
+     */
+    virtual Error BeginBatch() = 0;
+
+    /**
+     * Flushes the staged batch atomically in a single nft transaction.
+     *
+     * @return Error.
+     */
+    virtual Error FlushBatch() = 0;
+
+    /**
+     * Discards the staged batch and leaves batch mode without applying anything to the kernel
+     * (unlike FlushBatch which commits, or Revert which undoes an already-applied batch).
+     *
+     * @return Error.
+     */
+    virtual Error AbortBatch() = 0;
+
+    /**
+     * Reverts the flushed batch, deleting everything it applied by handle.
+     *
+     * @return Error.
+     */
+    virtual Error Revert() = 0;
 };
 
 /** @}*/