Move storage stats calculation to coro actor.

6 files changed, 106 insertions, 69 deletions

diff --git a/ydb/core/mind/bscontroller/storage_stats_calculator.cpp b/ydb/core/mind/bscontroller/storage_stats_calculator.cpp
index 83d655945c6..420471d3b44 100644
--- a/ydb/core/mind/bscontroller/storage_stats_calculator.cpp
+++ b/ydb/core/mind/bscontroller/storage_stats_calculator.cpp
@@ -5,38 +5,48 @@
 #include "impl.h"
 #include "sys_view.h"
 
+#include <ydb/core/blobstorage/base/utility.h>
+#include <ydb/core/protos/services.pb.h>
+
 #include <library/cpp/actors/core/actor.h>
+#include <library/cpp/actors/core/actor_coroutine.h>
+#include <library/cpp/actors/core/events.h>
+
+#include <util/generic/ptr.h>
+#include <util/system/yassert.h>
 
 #include <memory>
 #include <vector>
 
 namespace NKikimr::NBsController {
 
-class TStorageStatsCalculator : public TActor<TStorageStatsCalculator> {
-public:
-    TStorageStatsCalculator()
-        : TActor(&TStorageStatsCalculator::StateWork)
-    {}
-
-    STRICT_STFUNC(StateWork,
-        hFunc(TEvCalculateStorageStatsRequest, Handle);
-        cFunc(TEvents::TSystem::Poison, PassAway);
-    )
-
-    void Handle(TEvCalculateStorageStatsRequest::TPtr& ev) {
-        auto response = std::make_unique<TEvCalculateStorageStatsResponse>();
-        const auto& request = *(ev->Get());
-        response->StorageStats = GenerateStorageStats(request.SystemViewsState, request.HostRecordMap, request.GroupReserveMin, request.GroupReservePart);
-        Send(ev->Sender, response.release());
-    }
+/* TStorageStatsCoroCalculatorImpl */
 
+class TStorageStatsCoroCalculatorImpl : public TActorCoroImpl {
 private:
-    std::vector<NKikimrSysView::TStorageStatsEntry> GenerateStorageStats(
-            const TControllerSystemViewsState& systemViewsState,
-            const TBlobStorageController::THostRecordMap& hostRecordMap,
-            ui32 groupReserveMin,
-            ui32 groupReservePart)
+    enum {
+        EvResume = EventSpaceBegin(TEvents::ES_PRIVATE)
+    };
+
+public:
+    TStorageStatsCoroCalculatorImpl(
+        const TControllerSystemViewsState& systemViewsState,
+        const TBlobStorageController::THostRecordMap& hostRecordMap,
+        ui32 groupReserveMin,
+        ui32 groupReservePart)
+        : TActorCoroImpl(/* stackSize */ 64 * 1024, /* allowUnhandledPoisonPill */ true, /* allowUnhandledDtor */ true)
+        , SystemViewsState(systemViewsState)
+        , HostRecordMap(hostRecordMap)
+        , GroupReserveMin(groupReserveMin)
+        , GroupReservePart(groupReservePart)
     {
+    }
+
+    void ProcessUnexpectedEvent(TAutoPtr<IEventHandle> ev) override {
+        Y_FAIL("unexpected event Type# 0x%08" PRIx32, ev->GetTypeRewrite());
+    }
+
+    void Run() override {
         std::vector<NKikimrSysView::TStorageStatsEntry> storageStats;
 
         using TEntityKey = std::tuple<TString, TString>; // PDiskFilter, ErasureSpecies
@@ -59,7 +69,7 @@ private:
             }
         }
 
-        for (const auto& [key, value] : systemViewsState.StoragePools) {
+        for (const auto& [key, value] : SystemViewsState.StoragePools) {
             TEntityKey entityKey(value.GetPDiskFilter(), value.GetErasureSpeciesV2());
             const size_t index = entityMap.try_emplace(entityKey, storageStats.size()).first->second;
             if (index == storageStats.size()) {
@@ -77,7 +87,7 @@ private:
             spToEntity[key] = index;
         }
 
-        for (const auto& [groupId, group] : systemViewsState.Groups) {
+        for (const auto& [groupId, group] : SystemViewsState.Groups) {
             const TBlobStorageController::TBoxStoragePoolId key(group.GetBoxId(), group.GetStoragePoolId());
             if (const auto it = spToEntity.find(key); it != spToEntity.end()) {
                 auto& e = storageStats[it->second];
@@ -87,9 +97,9 @@ private:
             }
         }
 
-        using T = std::decay_t<decltype(systemViewsState.PDisks)>::value_type;
+        using T = std::decay_t<decltype(SystemViewsState.PDisks)>::value_type;
         std::unordered_map<TBlobStorageController::TBoxId, std::vector<const T*>> boxes;
-        for (const auto& kv : systemViewsState.PDisks) {
+        for (const auto& kv : SystemViewsState.PDisks) {
             if (kv.second.HasBoxId()) {
                 boxes[kv.second.GetBoxId()].push_back(&kv);
             }
@@ -110,7 +120,7 @@ private:
                         const auto sharedWithOs = pdisk.HasSharedWithOs() ? MakeMaybe(pdisk.GetSharedWithOs()) : Nothing();
                         const auto readCentric = pdisk.HasReadCentric() ? MakeMaybe(pdisk.GetReadCentric()) : Nothing();
                         if (filter.MatchPDisk(pdisk.GetCategory(), sharedWithOs, readCentric)) {
-                            const TNodeLocation& location = hostRecordMap->GetLocation(pdiskId.NodeId);
+                            const TNodeLocation& location = HostRecordMap->GetLocation(pdiskId.NodeId);
                             const bool ok = mapper.RegisterPDisk({
                                 .PDiskId = pdiskId,
                                 .Location = location,
@@ -140,7 +150,7 @@ private:
                     for (const auto& realm : group) {
                         for (const auto& domain : realm) {
                             for (const auto& pdiskId : domain) {
-                                if (const auto it = systemViewsState.PDisks.find(pdiskId); it != systemViewsState.PDisks.end()) {
+                                if (const auto it = SystemViewsState.PDisks.find(pdiskId); it != SystemViewsState.PDisks.end()) {
                                     const NKikimrSysView::TPDiskInfo& pdisk = it->second;
                                     auto& pm = *pdiskMetrics.emplace(pdiskMetrics.end());
                                     auto& vm = *vdiskMetrics.emplace(vdiskMetrics.end());
@@ -162,14 +172,16 @@ private:
                     groupSizes.push_back(groupInfo.GetAvailableSize());
 
                     group.clear();
+
+                    Yield();
                 }
 
                 std::sort(groupSizes.begin(), groupSizes.end());
 
                 // adjust it according to reserve
                 const ui32 total = groupSizes.size() + entry.GetCurrentGroupsCreated();
-                ui32 reserve = groupReserveMin;
-                while (reserve < groupSizes.size() && (reserve - groupReserveMin) * 1000000 / total < groupReservePart) {
+                ui32 reserve = GroupReserveMin;
+                while (reserve < groupSizes.size() && (reserve - GroupReserveMin) * 1000000 / total < GroupReservePart) {
                     ++reserve;
                 }
                 reserve = Min<ui32>(reserve, groupSizes.size());
@@ -191,12 +203,35 @@ private:
             }
         }
 
-        return storageStats;
+        Send(ParentActorId, new TEvCalculateStorageStatsResponse(std::move(storageStats)));
+    }
+
+private:
+    void Yield() {
+        Send(new IEventHandle(EvResume, 0, SelfActorId, {}, nullptr, 0));
+        WaitForSpecificEvent([](IEventHandle& ev) { return ev.Type == EvResume; });
     }
+
+private:
+    TControllerSystemViewsState SystemViewsState;
+    TBlobStorageController::THostRecordMap HostRecordMap;
+    ui32 GroupReserveMin = 0;
+    ui32 GroupReservePart = 0;
 };
 
-IActor *CreateStorageStatsCalculator() {
-    return new TStorageStatsCalculator();
+std::unique_ptr<IActor> CreateStorageStatsCoroCalculator(
+    const TControllerSystemViewsState& systemViewsState,
+    const TBlobStorageController::THostRecordMap& hostRecordMap,
+    ui32 groupReserveMin,
+    ui32 groupReservePart)
+{
+    auto coroCalculatorImpl = MakeHolder<TStorageStatsCoroCalculatorImpl>(
+        systemViewsState,
+        hostRecordMap,
+        groupReserveMin,
+        groupReservePart);
+
+    return std::make_unique<TActorCoro>(std::move(coroCalculatorImpl), NKikimrServices::TActivity::BS_STORAGE_STATS_ACTOR);
 }
 
 } // NKikimr::NBsController
diff --git a/ydb/core/mind/bscontroller/storage_stats_calculator.h b/ydb/core/mind/bscontroller/storage_stats_calculator.h
index 3c9c57332fa..deae03ab513 100644
--- a/ydb/core/mind/bscontroller/storage_stats_calculator.h
+++ b/ydb/core/mind/bscontroller/storage_stats_calculator.h
@@ -1,9 +1,22 @@
 #pragma once
 
+#include "impl.h"
+
 #include <library/cpp/actors/core/actor.h>
 
+#include <util/system/types.h>
+
+#include <memory>
+#include <vector>
+
 namespace NKikimr::NBsController {
 
-NActors::IActor *CreateStorageStatsCalculator();
+struct TControllerSystemViewsState;
+
+std::unique_ptr<NActors::IActor> CreateStorageStatsCoroCalculator(
+    const TControllerSystemViewsState& state,
+    const TBlobStorageController::THostRecordMap& hostRecordMap,
+    ui32 groupReserveMin,
+    ui32 groupReservePart);
 
 } // NKikimr::NBsController
diff --git a/ydb/core/mind/bscontroller/sys_view.cpp b/ydb/core/mind/bscontroller/sys_view.cpp
index 3f828974a5f..e3522770b3a 100644
--- a/ydb/core/mind/bscontroller/sys_view.cpp
+++ b/ydb/core/mind/bscontroller/sys_view.cpp
@@ -86,7 +86,6 @@ class TSystemViewsCollector : public TActorBootstrapped<TSystemViewsCollector> {
     std::unordered_set<std::tuple<TString>> PDiskFilterCounters;
     std::unordered_set<std::tuple<TString, TString>> ErasureCounters;
     TActorId StorageStatsCalculatorId;
-    bool IsCalculateStorageStatsRequestInProgress = false;
 
 public:
     static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
@@ -101,11 +100,9 @@ public:
         Counters->RemoveSubgroup("subsystem", "storage_stats");
     }
 
-    void Bootstrap(const TActorContext& ctx) {
-        StorageStatsCalculatorId = RunInBatchPool(ctx, CreateStorageStatsCalculator());
+    void Bootstrap(const TActorContext&) {
         Become(&TThis::StateWork);
-
-        Schedule(TDuration::Seconds(1), new TEvScheduleCalculateStorageStatsRequest());
+        Schedule(TDuration::Seconds(0), new TEvCalculateStorageStatsRequest());
     }
 
     STRICT_STFUNC(StateWork,
@@ -115,7 +112,7 @@ public:
         hFunc(TEvSysView::TEvGetGroupsRequest, Handle);
         hFunc(TEvSysView::TEvGetStoragePoolsRequest, Handle);
         hFunc(TEvSysView::TEvGetStorageStatsRequest, Handle);
-        hFunc(TEvScheduleCalculateStorageStatsRequest, Handle);
+        hFunc(TEvCalculateStorageStatsRequest, Handle);
         hFunc(TEvCalculateStorageStatsResponse, Handle);
         cFunc(TEvents::TSystem::Poison, PassAway);
     )
@@ -133,7 +130,10 @@ public:
     }
 
     void PassAway() override {
-        TActivationContext::Send(new IEventHandle(TEvents::TSystem::Poison, 0, StorageStatsCalculatorId, {}, nullptr, 0));
+        if (StorageStatsCalculatorId) {
+            TActivationContext::Send(new IEventHandle(TEvents::TSystem::Poison, 0, StorageStatsCalculatorId, {}, nullptr, 0));
+        }
+
         TActorBootstrapped::PassAway();
     }
 
@@ -221,17 +221,21 @@ public:
         Send(ev->Sender, response.release());
     }
 
-    void Handle(TEvScheduleCalculateStorageStatsRequest::TPtr&) {
-        if (IsCalculateStorageStatsRequestInProgress) {
+    void Handle(TEvCalculateStorageStatsRequest::TPtr&) {
+        if (StorageStatsCalculatorId) {
             return;
         }
 
-        IsCalculateStorageStatsRequestInProgress = true;
+        auto& ctx = TActivationContext::AsActorContext();
+        auto actor = CreateStorageStatsCoroCalculator(
+            State,
+            HostRecords,
+            GroupReserveMin,
+            GroupReservePart);
 
-        auto request = std::make_unique<TEvCalculateStorageStatsRequest>(State, HostRecords, GroupReserveMin, GroupReservePart);
-        Send(StorageStatsCalculatorId, request.release());
+        StorageStatsCalculatorId = RunInBatchPool(ctx, actor.release());
 
-        Schedule(TDuration::Minutes(10), new TEvScheduleCalculateStorageStatsRequest());
+        Schedule(TDuration::Minutes(10), new TEvCalculateStorageStatsRequest());
     }
 
     void Handle(TEvCalculateStorageStatsResponse::TPtr& ev) {
@@ -243,7 +247,7 @@ public:
             StorageStats.reset();
         }
 
-        IsCalculateStorageStatsRequestInProgress = false;
+        StorageStatsCalculatorId = TActorId();
     }
 
     void UpdateStorageStatsCounters(const std::vector<NKikimrSysView::TStorageStatsEntry>& storageStats) {
diff --git a/ydb/core/mind/bscontroller/sys_view.h b/ydb/core/mind/bscontroller/sys_view.h
index 9c9f700f9d7..0fa9fac124e 100644
--- a/ydb/core/mind/bscontroller/sys_view.h
+++ b/ydb/core/mind/bscontroller/sys_view.h
@@ -35,34 +35,19 @@ struct TEvControllerUpdateSystemViews :
 struct TEvCalculateStorageStatsRequest :
     TEventLocal<TEvCalculateStorageStatsRequest, NSysView::TEvSysView::EvCalculateStorageStatsRequest>
 {
-    TEvCalculateStorageStatsRequest(
-        const TControllerSystemViewsState& systemViewsState,
-        const TBlobStorageController::THostRecordMap& hostRecordMap,
-        ui32 groupReserveMin,
-        ui32 groupReservePart)
-        : SystemViewsState(systemViewsState)
-        , HostRecordMap(hostRecordMap)
-        , GroupReserveMin(groupReserveMin)
-        , GroupReservePart(groupReservePart)
-    {
-    }
-
-    TControllerSystemViewsState SystemViewsState;
-    TBlobStorageController::THostRecordMap HostRecordMap;
-    ui32 GroupReserveMin;
-    ui32 GroupReservePart;
 };
 
 struct TEvCalculateStorageStatsResponse :
     TEventLocal<TEvCalculateStorageStatsResponse, NSysView::TEvSysView::EvCalculateStorageStatsResponse>
 {
+    template <typename T>
+    TEvCalculateStorageStatsResponse(T&& t)
+        : StorageStats(std::forward<T>(t))
+    {}
+
     std::vector<NKikimrSysView::TStorageStatsEntry> StorageStats;
 };
 
-struct TEvScheduleCalculateStorageStatsRequest :
-    TEventLocal<TEvScheduleCalculateStorageStatsRequest, NSysView::TEvSysView::EvScheduleCalculateStorageStatsRequest>
-{};
-
 struct TGroupDiskInfo {
     const NKikimrBlobStorage::TPDiskMetrics *PDiskMetrics;
     const NKikimrBlobStorage::TVDiskMetrics *VDiskMetrics;
diff --git a/ydb/core/protos/services.proto b/ydb/core/protos/services.proto
index 44c76870dc4..f745bea8f02 100644
--- a/ydb/core/protos/services.proto
+++ b/ydb/core/protos/services.proto
@@ -896,5 +896,6 @@ message TActivity {
         BLOB_DEPOT_ACTOR = 570;
         BLOB_DEPOT_AGENT = 571;
         KQP_STREAM_LOOKUP_ACTOR = 572;
+        BS_STORAGE_STATS_ACTOR = 573;
     };
 };
diff --git a/ydb/core/sys_view/common/events.h b/ydb/core/sys_view/common/events.h
index 3d363b953e6..fb4347db616 100644
--- a/ydb/core/sys_view/common/events.h
+++ b/ydb/core/sys_view/common/events.h
@@ -74,7 +74,6 @@ struct TEvSysView {
 
         EvCalculateStorageStatsRequest,
         EvCalculateStorageStatsResponse,
-        EvScheduleCalculateStorageStatsRequest,
 
         EvEnd,
     };