Merge 844d30b into 34e573c

Author: robdrynkin

ydb/core/blobstorage/vdisk/balance/balancing_actor.cpp

@@ -87,6 +87,9 @@ namespace NBalancing {
         ///////////////////////////////////////////////////////////////////////////////////////////
 
         void ContinueBalancing() {
+            Ctx->MonGroup.PlannedToSendOnMain() = SendOnMainParts.Data.size();
+            Ctx->MonGroup.CandidatesToDelete() = TryDeleteParts.Data.size();
+
             if (SendOnMainParts.Empty() && TryDeleteParts.Empty()) {
                 // no more parts to send or delete
                 STLOG(PRI_INFO, BS_VDISK_BALANCING, BSVB03, VDISKP(Ctx->VCtx, "Balancing completed"));
@@ -101,8 +104,6 @@ namespace NBalancing {
 
         void ScheduleJobQuant() {
             Ctx->MonGroup.ReplTokenAquired()++;
-            Ctx->MonGroup.PlannedToSendOnMain() = SendOnMainParts.Data.size();
-            Ctx->MonGroup.CandidatesToDelete() = TryDeleteParts.Data.size();
 
             // once repl token received, start balancing - waking up sender and deleter
             STLOG(PRI_INFO, BS_VDISK_BALANCING, BSVB02, VDISKP(Ctx->VCtx, "Schedule job quant"),
@@ -128,16 +129,21 @@ namespace NBalancing {
             THPTimer timer;
 
             for (ui32 cnt = 0; It.Valid(); It.Next(), ++cnt) {
-                if (cnt % 100 == 99 && TDuration::Seconds(timer.Passed()) > JOB_GRANULARITY) {
+                if (cnt % 128 == 127 && TDuration::Seconds(timer.Passed()) > JOB_GRANULARITY) {
                     // actor should not block the thread for a long time, so we should yield
-                    // STLOG(PRI_DEBUG, BS_VDISK_BALANCING, BSVB04, VDISKP(Ctx->VCtx, "Collect keys"), (collected, cnt), (passed, timer.Passed()));
+                    STLOG(PRI_DEBUG, BS_VDISK_BALANCING, BSVB04, VDISKP(Ctx->VCtx, "Collect keys"), (collected, cnt), (passed, timer.Passed()));
                     Send(SelfId(), new NActors::TEvents::TEvWakeup());
                     return;
                 }
 
                 const auto& top = GInfo->GetTopology();
                 const auto& key = It.GetCurKey().LogoBlobID();
 
+                if (BALANCE_ONLY_HUGE_BLOBS && !Ctx->HugeBlobCtx->IsHugeBlob(GInfo->Type, key, Ctx->MinREALHugeBlobInBytes)) {
+                    // skip non huge blobs
+                    continue;
+                }
+
                 TPartsCollectorMerger merger(top.GType);
                 It.PutToMerger(&merger);
 
@@ -197,6 +203,7 @@ namespace NBalancing {
                 Send(MakeBlobStorageReplBrokerID(), new TEvReleaseReplToken);
                 STLOG(PRI_INFO, BS_VDISK_BALANCING, BSVB04, VDISKP(Ctx->VCtx, "Epoch timeout"));
                 PassAway();
+                return;
             }
 
             if (BatchManager.IsBatchCompleted()) {
@@ -242,8 +249,8 @@ namespace NBalancing {
 
         void Handle(NActors::TEvents::TEvUndelivered::TPtr ev) {
             if (ev.Get()->Type == TEvReplToken::EventType) {
-                STLOG(PRI_WARN, BS_VDISK_BALANCING, BSVB06, VDISKP(Ctx->VCtx, "Ask repl token msg not delivered"), (SelfId, SelfId()), (PDiskId, Ctx->VDiskCfg->BaseInfo.PDiskId));
-                ScheduleJobQuant();
+                STLOG(PRI_ERROR, BS_VDISK_BALANCING, BSVB06, VDISKP(Ctx->VCtx, "Ask repl token msg not delivered"), (SelfId, SelfId()), (PDiskId, Ctx->VDiskCfg->BaseInfo.PDiskId));
+                ContinueBalancing();
             }
         }
 

ydb/core/blobstorage/vdisk/balance/defs.h

@@ -76,6 +76,8 @@ namespace NBalancing {
     static constexpr ui32 MAX_TO_DELETE_PER_EPOCH = 1000;
     static constexpr TDuration EPOCH_TIMEOUT = TDuration::Minutes(1);
 
+    static constexpr bool BALANCE_ONLY_HUGE_BLOBS = true;
+
 
     struct TEvBalancingSendPartsOnMain : TEventLocal<TEvBalancingSendPartsOnMain, TEvBlobStorage::EvBalancingSendPartsOnMain> {
         TEvBalancingSendPartsOnMain(const TVector<TLogoBlobID>& ids)