ensure boot queue progress (#16713)

Author: vporyadke

ydb/core/mind/hive/boot_queue.cpp

@@ -17,20 +17,54 @@ void TBootQueue::AddToBootQueue(TBootQueueRecord record) {
 }
 
 TBootQueue::TBootQueueRecord TBootQueue::PopFromBootQueue() {
-    TBootQueueRecord record = BootQueue.top();
-    BootQueue.pop();
+    TQueue& currentQueue = GetCurrentQueue();
+    TBootQueueRecord record = currentQueue.top();
+    currentQueue.pop();
+    if (ProcessWaitQueue) {
+        NextFromWaitQueue = !NextFromWaitQueue;
+    }
     return record;
 }
 
 void TBootQueue::AddToWaitQueue(TBootQueueRecord record) {
-    WaitQueue.emplace_back(record);
+    WaitQueue.push(record);
+}
+
+void TBootQueue::IncludeWaitQueue() {
+    ProcessWaitQueue = true;
 }
 
-void TBootQueue::MoveFromWaitQueueToBootQueue() {
-    for (TBootQueueRecord record : WaitQueue) {
-        AddToBootQueue(record);
+void TBootQueue::ExcludeWaitQueue() {
+    ProcessWaitQueue = false;
+}
+
+bool TBootQueue::Empty() const {
+    if (ProcessWaitQueue) {
+        return BootQueue.empty() && WaitQueue.empty();
+    } else {
+        return BootQueue.empty();
+    }
+}
+
+size_t TBootQueue::Size() const {
+    if (ProcessWaitQueue) {
+        return BootQueue.size() + WaitQueue.size();
+    } else {
+        return BootQueue.size();
+    }
+}
+
+TBootQueue::TQueue& TBootQueue::GetCurrentQueue() {
+    if (BootQueue.empty()) {
+        return WaitQueue;
+    }
+    if (WaitQueue.empty()) {
+        return BootQueue;
+    }
+    if (ProcessWaitQueue && NextFromWaitQueue) {
+        return WaitQueue;
     }
-    WaitQueue.clear();
+    return BootQueue;
 }
 
 }

ydb/core/mind/hive/boot_queue.h

@@ -52,18 +52,30 @@ struct TBootQueue {
 
     static_assert(sizeof(TBootQueueRecord) <= 24);
 
-    std::priority_queue<TBootQueueRecord, std::vector<TBootQueueRecord>> BootQueue;
-    std::deque<TBootQueueRecord> WaitQueue; // tablets from BootQueue waiting for new nodes
+    using TQueue = TPriorityQueue<TBootQueueRecord>;
 
+    TQueue BootQueue;
+    TQueue WaitQueue; // tablets from BootQueue waiting for new nodes
+private:
+    bool ProcessWaitQueue = false;
+    bool NextFromWaitQueue = false;
+
+public:
     void AddToBootQueue(TBootQueueRecord record);
     TBootQueueRecord PopFromBootQueue();
     void AddToWaitQueue(TBootQueueRecord record);
-    void MoveFromWaitQueueToBootQueue();
+    void IncludeWaitQueue();
+    void ExcludeWaitQueue();
+    bool Empty() const;
+    size_t Size() const;
 
     template<typename... Args>
     void EmplaceToBootQueue(Args&&... args) {
         BootQueue.emplace(args...);
     }
+
+private:
+    TQueue& GetCurrentQueue();
 };
 
 }

ydb/core/mind/hive/hive_events.h

@@ -51,7 +51,9 @@ struct TEvPrivate {
         {}
     };
 
-    struct TEvProcessBootQueue : TEventLocal<TEvProcessBootQueue, EvProcessBootQueue> {};
+    struct TEvProcessBootQueue : TEventLocal<TEvProcessBootQueue, EvProcessBootQueue> {
+        bool ProcessWaitQueue = false; // Only for use in tests
+    };
 
     struct TEvPostponeProcessBootQueue : TEventLocal<TEvPostponeProcessBootQueue, EvPostponeProcessBootQueue> {};
 

ydb/core/mind/hive/hive_impl.cpp

@@ -223,15 +223,17 @@ void THive::ExecuteProcessBootQueue(NIceDb::TNiceDb&, TSideEffects& sideEffects)
     THPTimer bootQueueProcessingTimer;
     if (ProcessWaitQueueScheduled) {
         BLOG_D("Handle ProcessWaitQueue (size: " << BootQueue.WaitQueue.size() << ")");
-        BootQueue.MoveFromWaitQueueToBootQueue();
+        BootQueue.IncludeWaitQueue();
         ProcessWaitQueueScheduled = false;
     }
     ProcessBootQueueScheduled = false;
     ui64 processedItems = 0;
     ui64 tabletsStarted = 0;
     TInstant postponedStart;
     TStackVec<TBootQueue::TBootQueueRecord> delayedTablets;
-    while (!BootQueue.BootQueue.empty() && processedItems < GetMaxBootBatchSize()) {
+    std::vector<TBootQueue::TBootQueueRecord> waitingTablets;
+    waitingTablets.reserve(std::min<size_t>(BootQueue.Size(), GetMaxBootBatchSize()));
+    while (!BootQueue.Empty() && processedItems < GetMaxBootBatchSize()) {
         TBootQueue::TBootQueueRecord record = BootQueue.PopFromBootQueue();
         BLOG_TRACE("Tablet " << record.TabletId << "." << record.FollowerId << " has priority " << record.Priority);
         ++processedItems;
@@ -260,7 +262,7 @@ void THive::ExecuteProcessBootQueue(NIceDb::TNiceDb&, TSideEffects& sideEffects)
                         sideEffects.Send(actorToNotify, new TEvPrivate::TEvRestartComplete(tablet->GetFullTabletId(), "boot delay"));
                     }
                     tablet->ActorsToNotifyOnRestart.clear();
-                    BootQueue.AddToWaitQueue(record); // waiting for new node
+                    waitingTablets.push_back(record); // waiting for new node
                     tablet->InWaitQueue = true;
                     continue;
                 }
@@ -280,10 +282,16 @@ void THive::ExecuteProcessBootQueue(NIceDb::TNiceDb&, TSideEffects& sideEffects)
             delayedTablets.push_back(record);
         }
     }
+    if (waitingTablets.size() == processedItems || BootQueue.WaitQueue.empty()) {
+        BootQueue.ExcludeWaitQueue();
+    }
     for (TBootQueue::TBootQueueRecord record : delayedTablets) {
         record.Priority -= 1;
         BootQueue.AddToBootQueue(record);
     }
+    for (auto& record : waitingTablets) {
+        BootQueue.AddToWaitQueue(record);
+    }
     if (TabletCounters != nullptr) {
         UpdateCounterBootQueueSize(BootQueue.BootQueue.size());
         TabletCounters->Simple()[NHive::COUNTER_WAITQUEUE_SIZE].Set(BootQueue.WaitQueue.size());
@@ -301,7 +309,7 @@ void THive::ExecuteProcessBootQueue(NIceDb::TNiceDb&, TSideEffects& sideEffects)
             BLOG_D("ProcessBootQueue - BootQueue throttling (size: " << BootQueue.BootQueue.size() << ")");
             return;
         }
-        if (processedItems == GetMaxBootBatchSize()) {
+        if (processedItems == GetMaxBootBatchSize() && !BootQueue.Empty()) {
             BLOG_D("ProcessBootQueue - rescheduling");
             ProcessBootQueue();
         } else if (postponedStart > now) {
@@ -316,8 +324,11 @@ void THive::HandleInit(TEvPrivate::TEvProcessBootQueue::TPtr&) {
     Schedule(TDuration::Seconds(1), new TEvPrivate::TEvProcessBootQueue());
 }
 
-void THive::Handle(TEvPrivate::TEvProcessBootQueue::TPtr&) {
+void THive::Handle(TEvPrivate::TEvProcessBootQueue::TPtr& ev) {
     BLOG_TRACE("ProcessBootQueue - executing");
+    if (ev->Get()->ProcessWaitQueue) {
+        ProcessWaitQueue();
+    }
     Execute(CreateProcessBootQueue());
 }
 
@@ -2240,14 +2251,14 @@ TResourceRawValues THive::GetDefaultResourceInitialMaximumValues() {
 }
 
 void THive::ProcessTabletBalancer() {
-    if (!ProcessTabletBalancerScheduled && !ProcessTabletBalancerPostponed && BootQueue.BootQueue.empty()) {
+    if (!ProcessTabletBalancerScheduled && !ProcessTabletBalancerPostponed && BootQueue.Empty()) {
         Schedule(GetBalancerCooldown(LastBalancerTrigger), new TEvPrivate::TEvProcessTabletBalancer());
         ProcessTabletBalancerScheduled = true;
     }
 }
 
 void THive::ProcessStorageBalancer() {
-    if (!ProcessStorageBalancerScheduled && BootQueue.BootQueue.empty()) {
+    if (!ProcessStorageBalancerScheduled && BootQueue.Empty()) {
         Schedule(GetBalancerCooldown(EBalancerType::Storage), new TEvPrivate::TEvProcessStorageBalancer());
         ProcessStorageBalancerScheduled = true;
     }
@@ -2484,10 +2495,7 @@ void THive::UpdateTotalResourceValues(
     TInstant now = TInstant::Now();
 
     if (LastResourceChangeReaction + GetResourceChangeReactionPeriod() < now) {
-        // in case we had overloaded nodes
-        if (!BootQueue.WaitQueue.empty()) {
-            ProcessWaitQueue();
-        } else if (!BootQueue.BootQueue.empty()) {
+        if (!BootQueue.Empty()) {
             ProcessBootQueue();
         }
         ProcessTabletBalancer();

ydb/core/mind/hive/hive_impl_ut.cpp

@@ -58,14 +58,29 @@ Y_UNIT_TEST_SUITE(THiveImplTest) {
         timer.Reset();
 
         double maxP = 100;
+        std::vector<TBootQueue::TBootQueueRecord> records;
+        records.reserve(NUM_TABLETS);
+        unsigned i = 0;
 
-        while (!bootQueue.BootQueue.empty()) {
+        while (!bootQueue.Empty()) {
             auto record = bootQueue.PopFromBootQueue();
             UNIT_ASSERT(record.Priority <= maxP);
             maxP = record.Priority;
-            auto itTablet = tablets.find(record.TabletId);
-            if (itTablet != tablets.end()) {
-                bootQueue.AddToWaitQueue(itTablet->second);
+            UNIT_ASSERT(tablets.contains(record.TabletId));
+            records.push_back(record);
+            if (++i == NUM_TABLETS / 2) {
+                // to test both modes 
+                bootQueue.IncludeWaitQueue();
+            }
+        }
+        bootQueue.ExcludeWaitQueue();
+
+        i = 0;
+        for (auto& record : records) {
+            if (++i % 3 == 0) {
+                bootQueue.AddToBootQueue(record);
+            } else {
+                bootQueue.AddToWaitQueue(record);
             }
         }
 
@@ -81,7 +96,11 @@ Y_UNIT_TEST_SUITE(THiveImplTest) {
 
         timer.Reset();
 
-        bootQueue.MoveFromWaitQueueToBootQueue();
+        bootQueue.IncludeWaitQueue();
+        while (!bootQueue.Empty()) {
+            bootQueue.PopFromBootQueue();
+        }
+        bootQueue.ExcludeWaitQueue();
 
         passed = timer.Get().SecondsFloat();
         Ctest << "Move = " << passed << Endl;

ydb/core/mind/hive/hive_ut.cpp

@@ -634,6 +634,14 @@ Y_UNIT_TEST_SUITE(THiveTest) {
         runtime.SendToPipe(hiveTablet, senderB, new TEvHive::TEvReassignTabletSpace(tabletId, channels), 0, GetPipeConfigWithRetries());
     }
 
+    TActorId GetHiveActor(TTestActorRuntime& runtime, ui64 hiveTablet) {
+        TActorId senderB = runtime.AllocateEdgeActor(0);
+        runtime.SendToPipe(hiveTablet, senderB, new TEvHive::TEvTabletMetrics, 0, GetPipeConfigWithRetries());
+        TAutoPtr<IEventHandle> handle;
+        runtime.GrabEdgeEventRethrow<TEvLocal::TEvTabletMetricsAck>(handle);
+        return handle->Sender;
+    }
+
     void MakeSureTabletIsDown(TTestActorRuntime &runtime, ui64 tabletId, ui32 nodeIndex) {
         TActorId sender = runtime.AllocateEdgeActor(nodeIndex);
         runtime.ConnectToPipe(tabletId, sender, nodeIndex, NTabletPipe::TClientConfig());
@@ -3475,7 +3483,8 @@ Y_UNIT_TEST_SUITE(THiveTest) {
         TVector<ui64> tabletIds;
         const ui64 hiveTablet = MakeDefaultHiveID();
         const ui64 testerTablet = MakeTabletID(false, 1);
-        CreateTestBootstrapper(runtime, CreateTestTabletInfo(hiveTablet, TTabletTypes::Hive), &CreateDefaultHive);
+        const TActorId bootstrapper = CreateTestBootstrapper(runtime, CreateTestTabletInfo(hiveTablet, TTabletTypes::Hive), &CreateDefaultHive);
+        runtime.EnableScheduleForActor(bootstrapper);
         {
             TDispatchOptions options;
             options.FinalEvents.emplace_back(TEvLocal::EvSyncTablets, runtime.GetNodeCount());
@@ -7039,6 +7048,43 @@ Y_UNIT_TEST_SUITE(THiveTest) {
         }
     }
 
+    Y_UNIT_TEST(TestBootProgress) {
+        TTestBasicRuntime runtime(1, false);
+        Setup(runtime, true, 3, [](TAppPrepare& app) {
+            app.HiveConfig.SetMaxBootBatchSize(1);
+            app.HiveConfig.SetResourceChangeReactionPeriod(0);
+        });
+        const ui64 hiveTablet = MakeDefaultHiveID();
+        const ui64 testerTablet = MakeTabletID(false, 1);
+        const TActorId bootstrapper = CreateTestBootstrapper(runtime, CreateTestTabletInfo(hiveTablet, TTabletTypes::Hive), &CreateDefaultHive);
+        runtime.EnableScheduleForActor(bootstrapper);
+        {
+            TDispatchOptions options;
+            options.FinalEvents.emplace_back(TEvLocal::EvSyncTablets, runtime.GetNodeCount());
+            runtime.DispatchEvents(options, TDuration::Zero());
+        }
+        for (int i = 0; i < 5; ++i) {
+            THolder<TEvHive::TEvCreateTablet> ev(new TEvHive::TEvCreateTablet(testerTablet, 100500 + i, TTabletTypes::Hive, BINDED_CHANNELS));
+            ev->Record.AddAllowedDomains();
+            ev->Record.MutableAllowedDomains(0)->SetSchemeShard(52); // garbage domain id - these tablets will never boot
+            ev->Record.MutableAllowedDomains(0)->SetPathId(42);
+            SendCreateTestTablet(runtime, hiveTablet, testerTablet, std::move(ev), 0, false);
+        }
+        TActorId hiveActor = GetHiveActor(runtime, hiveTablet);
+        // Simulate a situation when wait queue is constantly processed
+        // this could happen e. g. when nodes are often restarting
+        // (previously it would happen all the time because of metric updates)
+        auto handler = runtime.AddObserver<NHive::TEvPrivate::TEvProcessBootQueue>([=](auto&& ev) {
+            if (ev->Recipient == hiveActor) {
+                ev->Get()->ProcessWaitQueue = true;
+            }
+        });
+        THolder<TEvHive::TEvCreateTablet> ev(new TEvHive::TEvCreateTablet(testerTablet, 100505, TTabletTypes::Dummy, BINDED_CHANNELS));
+        auto tabletId = SendCreateTestTablet(runtime, hiveTablet, testerTablet, std::move(ev), 0, false);
+        MakeSureTabletIsUp(runtime, tabletId, 0);
+
+    }
+
     Y_UNIT_TEST(TestStopTenant) {
         TTestBasicRuntime runtime(2, false);
         Setup(runtime, true);

ydb/core/mind/hive/monitoring.cpp

@@ -245,7 +245,7 @@ class TTxMonEvent_MemStateTablets : public TTransactionBase<THive> {
         }
         if (WaitingOnly) {
             tabletIdIndex.reserve(Self->BootQueue.WaitQueue.size());
-            for (const TBootQueue::TBootQueueRecord& rec : Self->BootQueue.WaitQueue) {
+            for (const TBootQueue::TBootQueueRecord& rec : Self->BootQueue.WaitQueue.Container()) {
                 TTabletInfo* tablet = Self->FindTablet(rec.TabletId, rec.FollowerId);
                 if (tablet != nullptr) {
                     tabletIdIndex.push_back({tabletIndexFunction(*tablet), tablet});