optimize initializiton of allocation units (#10969)

Author: vporyadke

ydb/core/mind/hive/hive_events.h

@@ -34,6 +34,7 @@ struct TEvPrivate {
         EvDeleteNode,
         EvCanMoveTablets,
         EvUpdateDataCenterFollowers,
+        EvGenerateTestData,
         EvRefreshScaleRecommendation,
         EvEnd
     };
@@ -129,6 +130,8 @@ struct TEvPrivate {
         TEvUpdateDataCenterFollowers(TDataCenterId dataCenter) : DataCenter(dataCenter) {};
     };
 
+    struct TEvGenerateTestData : TEventLocal<TEvGenerateTestData, EvGenerateTestData> {};
+  
     struct TEvRefreshScaleRecommendation : TEventLocal<TEvRefreshScaleRecommendation, EvRefreshScaleRecommendation> {};
 };
 

ydb/core/mind/hive/hive_impl.cpp

@@ -3064,7 +3064,8 @@ void THive::ProcessEvent(std::unique_ptr<IEventHandle> event) {
         hFunc(TEvHive::TEvRequestTabletDistribution, Handle);
         hFunc(TEvPrivate::TEvUpdateDataCenterFollowers, Handle);
         hFunc(TEvHive::TEvRequestScaleRecommendation, Handle);
-        hFunc(TEvPrivate::TEvRefreshScaleRecommendation, Handle)
+        hFunc(TEvPrivate::TEvGenerateTestData, Handle);
+        hFunc(TEvPrivate::TEvRefreshScaleRecommendation, Handle);
     }
 }
 
@@ -3169,6 +3170,7 @@ STFUNC(THive::StateWork) {
         fFunc(TEvHive::TEvRequestTabletDistribution::EventType, EnqueueIncomingEvent);
         fFunc(TEvPrivate::TEvUpdateDataCenterFollowers::EventType, EnqueueIncomingEvent);
         fFunc(TEvHive::TEvRequestScaleRecommendation::EventType, EnqueueIncomingEvent);
+        fFunc(TEvPrivate::TEvGenerateTestData::EventType, EnqueueIncomingEvent);
         fFunc(TEvPrivate::TEvRefreshScaleRecommendation::EventType, EnqueueIncomingEvent);
         hFunc(TEvPrivate::TEvProcessIncomingEvent, Handle);
     default:
@@ -3628,6 +3630,12 @@ void THive::Handle(TEvHive::TEvRequestScaleRecommendation::TPtr& ev) {
     Send(ev->Sender, response.release());
 }
 
+void THive::Handle(TEvPrivate::TEvGenerateTestData::TPtr&) {
+    for (int i = 0; i < 4; ++i) {
+        Execute(CreateGenerateTestData(i));
+    }
+}
+
 TVector<TNodeId> THive::GetNodesForWhiteboardBroadcast(size_t maxNodesToReturn) {
     TVector<TNodeId> nodes;
     TNodeId selfNodeId = SelfId().NodeId();

ydb/core/mind/hive/hive_impl.h

@@ -304,6 +304,7 @@ class THive : public TActor<THive>, public TTabletExecutedFlat, public THiveShar
     ITransaction* CreateUpdateTabletsObject(TEvHive::TEvUpdateTabletsObject::TPtr event);
     ITransaction* CreateUpdateDomain(TSubDomainKey subdomainKey, TEvHive::TEvUpdateDomain::TPtr event = {});
     ITransaction* CreateUpdateDcFollowers(const TDataCenterId& dc);
+    ITransaction* CreateGenerateTestData(uint64_t seed);
     ITransaction* CreateDeleteNode(TNodeId nodeId);
 
 public:
@@ -580,6 +581,7 @@ class THive : public TActor<THive>, public TTabletExecutedFlat, public THiveShar
     void Handle(TEvHive::TEvRequestTabletDistribution::TPtr& ev);
     void Handle(TEvPrivate::TEvUpdateDataCenterFollowers::TPtr& ev);
     void Handle(TEvHive::TEvRequestScaleRecommendation::TPtr& ev);
+    void Handle(TEvPrivate::TEvGenerateTestData::TPtr& ev);
     void Handle(TEvPrivate::TEvRefreshScaleRecommendation::TPtr& ev);
 
 protected:

ydb/core/mind/hive/hive_ut.cpp

@@ -60,7 +60,7 @@ namespace {
     using namespace NActors;
 
     void SetupLogging(TTestActorRuntime& runtime) {
-        NActors::NLog::EPriority priority = ENABLE_DETAILED_HIVE_LOG ? NLog::PRI_DEBUG : NLog::PRI_ERROR;
+        NActors::NLog::EPriority priority = ENABLE_DETAILED_HIVE_LOG ? NLog::PRI_DEBUG : NLog::PRI_NOTICE;
         NActors::NLog::EPriority otherPriority = NLog::PRI_DEBUG;
 
         if (ENABLE_DETAILED_HIVE_LOG) {
@@ -6841,6 +6841,32 @@ Y_UNIT_TEST_SUITE(THiveTest) {
     }
 }
 
+Y_UNIT_TEST_SUITE(THeavyPerfTest) {
+    Y_UNIT_TEST(TTestLoadEverything) {
+        TTestBasicRuntime runtime(2, false);
+        Setup(runtime, true);
+        const ui64 hiveTablet = MakeDefaultHiveID();
+        CreateTestBootstrapper(runtime, CreateTestTabletInfo(hiveTablet, TTabletTypes::Hive), &CreateDefaultHive);
+        NTestSuiteTHiveTest::MakeSureTabletIsUp(runtime, hiveTablet, 0);
+        TActorId senderB = runtime.AllocateEdgeActor(0);
+        runtime.SendToPipe(hiveTablet, senderB, new NHive::TEvPrivate::TEvGenerateTestData(), 0, GetPipeConfigWithRetries());
+        {
+            TDispatchOptions options;
+            options.FinalEvents.emplace_back(TEvTablet::TEvCommit::EventType, 2);
+            runtime.DispatchEvents(options);
+        }
+
+        runtime.Register(CreateTabletKiller(hiveTablet));
+
+        {
+            TDispatchOptions options;
+            options.FinalEvents.emplace_back(NHive::TEvPrivate::EvBootTablets);
+            runtime.DispatchEvents(options);
+        }
+
+    }
+}
+
 Y_UNIT_TEST_SUITE(TStorageBalanceTest) {
     static constexpr i64 DEFAULT_BIND_SIZE = 100'000'000;
     const std::vector<TString> STORAGE_POOLS = {"def1"};

ydb/core/mind/hive/leader_tablet_info.cpp

@@ -201,8 +201,11 @@ TActorId TLeaderTabletInfo::SetLockedToActor(const TActorId& actor, const TDurat
 }
 
 void TLeaderTabletInfo::AcquireAllocationUnits() {
-    for (ui32 channel = 0; channel < TabletStorageInfo->Channels.size(); ++channel) {
-        AcquireAllocationUnit(channel);
+    for (const auto& channel : TabletStorageInfo->Channels) {
+        if (!channel.History.empty()) {
+            TStoragePoolInfo& storagePool = Hive.GetStoragePool(channel.StoragePool);
+            storagePool.AcquireAllocationUnit(this, channel.Channel, channel.History.back().GroupID);
+        }
     }
 }
 

ydb/core/mind/hive/leader_tablet_info.h

@@ -23,10 +23,9 @@ struct TTabletCategoryInfo {
 struct TStoragePoolInfo;
 
 struct TLeaderTabletInfo : TTabletInfo {
-protected:
+public:
     static TString DEFAULT_STORAGE_POOL_NAME;
 
-public:
     struct TChannel {
         TTabletId TabletId;
         ui32 ChannelId;

ydb/core/mind/hive/storage_group_info.h

@@ -3,6 +3,8 @@
 #include "hive.h"
 #include "leader_tablet_info.h"
 
+#include <library/cpp/containers/absl_flat_hash/flat_hash_set.h>
+
 namespace NKikimr {
 namespace NHive {
 
@@ -42,7 +44,7 @@ struct TChannelHash {
 struct TStorageGroupInfo {
     const TStoragePoolInfo& StoragePool;
     TStorageGroupId Id;
-    std::unordered_set<TLeaderTabletInfo::TChannel, TChannelHash> Units;
+    absl::flat_hash_set<TLeaderTabletInfo::TChannel, TChannelHash> Units;
     TStorageResources AcquiredResources;
     TStorageResources MaximumResources;
     NKikimrBlobStorage::TEvControllerSelectGroupsResult::TGroupParameters GroupParameters;

ydb/core/mind/hive/storage_pool_info.cpp

@@ -14,9 +14,10 @@ TStoragePoolInfo::TStoragePoolInfo(const TString& name, THiveSharedSettings* hiv
 }
 
 TStorageGroupInfo& TStoragePoolInfo::GetStorageGroup(TStorageGroupId groupId) {
-    auto it = Groups.find(groupId);
+    decltype(Groups)::insert_ctx ctx;
+    auto it = Groups.find(groupId, ctx);
     if (it == Groups.end()) {
-        it = Groups.emplace(std::piecewise_construct, std::tuple<TStorageGroupId>(groupId), std::tuple<const TStoragePoolInfo&, TStorageGroupId>(*this, groupId)).first;
+        it = Groups.emplace_direct(ctx, std::piecewise_construct, std::tuple<TStorageGroupId>(groupId), std::tuple<const TStoragePoolInfo&, TStorageGroupId>(*this, groupId));
     }
     return it->second;
 }

ydb/core/mind/hive/tx__generate_data_ut.cpp

@@ -0,0 +1,54 @@
+#include "hive_impl.h"
+#include "hive_log.h"
+
+namespace NKikimr {
+namespace NHive {
+
+class TTxGenerateTestData : public TTransactionBase<THive> {
+public:
+    TTxGenerateTestData(THive *hive, uint64_t seed)
+        : TBase(hive)
+        , Seed(seed)
+    {}
+
+    TTxType GetTxType() const override { return -1; }
+
+    const std::vector<TString> STORAGE_POOLS = {"def1", "def2", "def3", "def4", "def5", "def6", "def7"};
+    static constexpr size_t NUM_TABLETS = 250'000;
+    static constexpr size_t NUM_GROUPS = 30'000;
+
+    uint64_t Seed;
+
+    bool Execute(TTransactionContext &txc, const TActorContext&) override {
+        NIceDb::TNiceDb db(txc.DB);
+
+        std::mt19937 engine(Seed);
+        TTabletId nextTabletId = 0x10000 + NUM_TABLETS * Seed;
+        std::uniform_int_distribution<size_t> getNumChannels(1, 10);
+        std::uniform_int_distribution<TStorageGroupId> getGroup(1, NUM_GROUPS);
+
+
+        for (size_t i = 0; i < NUM_TABLETS; ++i) {
+            auto tabletId = nextTabletId++;
+            db.Table<Schema::Tablet>().Key(tabletId).Update<Schema::Tablet::KnownGeneration>(1);
+            auto numChannels = getNumChannels(engine);
+            for (size_t channel = 0; channel < numChannels; ++channel) {
+                auto groupId = getGroup(engine);
+                auto storagePool = STORAGE_POOLS[(groupId - 1) * STORAGE_POOLS.size() / NUM_GROUPS];
+                db.Table<Schema::TabletChannel>().Key(tabletId, channel).Update<Schema::TabletChannel::StoragePool>(storagePool);
+                db.Table<Schema::TabletChannelGen>().Key(tabletId, channel, 1).Update<Schema::TabletChannelGen::Group>(groupId);
+            }
+        }
+
+        return true;
+    }
+
+    void Complete(const TActorContext&) override {}
+};
+
+ITransaction* THive::CreateGenerateTestData(uint64_t seed) {
+    return new TTxGenerateTestData(this, seed);
+}
+
+} // NHive
+} // NKikimr

ydb/core/mind/hive/tx__load_everything.cpp

@@ -201,7 +201,7 @@ class TTxLoadEverything : public TTransactionBase<THive> {
             auto end = owners.GetValue<Schema::TabletOwners::End>();
             auto ownerId = owners.GetValue<Schema::TabletOwners::OwnerId>();
 
-            Self->Keeper.AddOwnedSequence(ownerId, {begin, end});
+           Self->Keeper.AddOwnedSequence(ownerId, {begin, end});
             if (!owners.Next()) {
                 return false;
             }
@@ -489,6 +489,9 @@ class TTxLoadEverything : public TTransactionBase<THive> {
                 if (tablet) {
                     ui32 channelId = tabletChannelRowset.GetValue<Schema::TabletChannel::Channel>();
                     TString storagePool = tabletChannelRowset.GetValue<Schema::TabletChannel::StoragePool>();
+                    if (storagePool.empty()) {
+                        storagePool = TLeaderTabletInfo::DEFAULT_STORAGE_POOL_NAME;
+                    }
                     Y_ABORT_UNLESS(tablet->BoundChannels.size() == channelId);
                     tablet->BoundChannels.emplace_back();
                     NKikimrStoragePool::TChannelBind& bind = tablet->BoundChannels.back();