Added milestone queue in replication to reduce work unit computations (#6561)

alexvru · web-flow · commit 0201d6625438 · 2024-07-11T18:42:38.000+03:00
diff --git a/ydb/core/blobstorage/vdisk/repl/blobstorage_hullrepljob.cpp b/ydb/core/blobstorage/vdisk/repl/blobstorage_hullrepljob.cpp
@@ -26,13 +26,15 @@ namespace NKikimr {
         TLogoBlobID LastKey;
         bool Eof;
         std::deque<TLogoBlobID> DroppedBlobs;
+        TMilestoneQueue MilestoneQueue;
 
         TEvReplPlanFinished(std::unique_ptr<TRecoveryMachine>&& recoveryMachine, const TLogoBlobID& lastKey, bool eof,
-                std::deque<TLogoBlobID>&& droppedBlobs)
+                std::deque<TLogoBlobID>&& droppedBlobs, TMilestoneQueue&& milestoneQueue)
             : RecoveryMachine(std::move(recoveryMachine))
             , LastKey(lastKey)
             , Eof(eof)
             , DroppedBlobs(std::move(droppedBlobs))
+            , MilestoneQueue(std::move(milestoneQueue))
         {}
     };
 
@@ -52,6 +54,7 @@ namespace NKikimr {
         std::deque<TLogoBlobID> DroppedBlobs;
         ui64 QuantumBytes = 0;
         bool AddingTasks = true;
+        TMilestoneQueue MilestoneQueue;
 
     public:
         void Bootstrap(const TActorId& parentId) {
@@ -106,43 +109,83 @@ namespace NKikimr {
             } else {
                 // scan through the index until we have enough blobs to recover or the time is out
                 const TBlobStorageGroupInfo::TTopology& topology = *ReplCtx->VCtx->Top;
-                for (it.Seek(StartKey); it.Valid(); it.Next()) {
-                    StartKey = it.GetCurKey().LogoBlobID();
+
+                if (StartKey == TLogoBlobID()) {
+                    it.SeekToFirst();
+                } else {
+                    it.Seek(StartKey);
+                    if (it.Valid() && it.GetCurKey() == StartKey) {
+                        it.Next();
+                    }
+                }
+
+                auto checkRestart = [&] {
                     if (++counter % 1024 == 0 && GetCycleCountFast() >= plannedEndTime) {
                         // we have event processing timer expired, restart processing later with new snapshot starting
                         // with current key
                         Send(ReplCtx->SkeletonId, new TEvTakeHullSnapshot(true));
-                        return;
-                    } else if (AddingTasks) {
+                        return true;
+                    }
+                    return false;
+                };
+
+                if (AddingTasks) {
+                    for (; it.Valid(); it.Next()) {
+                        if (checkRestart()) {
+                            return;
+                        }
+
+                        StartKey = it.GetCurKey().LogoBlobID();
+
                         // we still have some space in recovery machine logic, so we can add new item
                         ProcessItem(it, *barriers, allowKeepFlags);
-                    } else {
-                        // no space in recovery machine logic, but we still have to count remaining work
-                        const TMemRecLogoBlob memRec = it.GetMemRec();
-                        const TIngress ingress = memRec.GetIngress();
-                        const auto parts = ingress.PartsWeMustHaveLocally(&topology, ReplCtx->VCtx->ShortSelfVDisk,
-                            StartKey) - ingress.LocalParts(topology.GType);
-                        if (!parts.Empty() && barriers->Keep(StartKey, memRec, {}, allowKeepFlags,
-                                true /*allowGarbageCollection*/).KeepData) {
-                            ++ReplInfo->ItemsTotal;
-                            ReplInfo->WorkUnitsTotal += StartKey.BlobSize();
-                        }
+                        MilestoneQueue.PopIfNeeded(StartKey);
 
-                        if (!KeyToResumeNextTime) {
-                            // this is first valid key that is not processed with ProcessItem, so we remember it to
-                            // start next quantum with this exact key
-                            KeyToResumeNextTime.emplace(StartKey);
+                        if (!AddingTasks) { // we have finished adding tasks after this key, remember it
+                            it.Next();
+                            Y_ABORT_UNLESS(!KeyToResumeNextTime);
+                            if (it.Valid()) {
+                                KeyToResumeNextTime.emplace(it.GetCurKey().LogoBlobID());
+                            }
+                            break;
                         }
                     }
+                    eof = !it.Valid(); // we finish this quantum when there are no more tasks to generate
                 }
 
-                // we shall run next quantum only if we have KeyToResumeNextTime filled in
-                eof = !KeyToResumeNextTime;
+                for (; it.Valid(); it.Next()) {
+                    if (checkRestart()) {
+                        return;
+                    }
+
+                    StartKey = it.GetCurKey().LogoBlobID();
+
+                    // check the milestone queue, if we have requested blob
+                    if (MilestoneQueue.Match(StartKey, &ReplInfo->ItemsTotal, &ReplInfo->WorkUnitsTotal)) {
+                        break;
+                    }
+
+                    // no space in recovery machine logic, but we still have to count remaining work
+                    const TMemRecLogoBlob memRec = it.GetMemRec();
+                    const TIngress ingress = memRec.GetIngress();
+                    const auto parts = ingress.PartsWeMustHaveLocally(&topology, ReplCtx->VCtx->ShortSelfVDisk,
+                        StartKey) - ingress.LocalParts(topology.GType);
+                    if (!parts.Empty() && barriers->Keep(StartKey, memRec, {}, allowKeepFlags,
+                            true /*allowGarbageCollection*/).KeepData) {
+                        ++ReplInfo->ItemsTotal;
+                        ReplInfo->WorkUnitsTotal += StartKey.BlobSize();
+                        MilestoneQueue.Push(StartKey, StartKey.BlobSize());
+                    }
+                }
+
+                if (!it.Valid()) {
+                    MilestoneQueue.Finish();
+                }
             }
 
             // the planning stage has finished, issue reply to the job actor
             Send(Recipient, new TEvReplPlanFinished(std::move(RecoveryMachine), KeyToResumeNextTime.value_or(TLogoBlobID()),
-                eof, std::move(DroppedBlobs)));
+                eof, std::move(DroppedBlobs), std::move(MilestoneQueue)));
 
             // finish processing for this actor
             PassAway();
@@ -219,13 +262,15 @@ namespace NKikimr {
                 const TLogoBlobID &startKey,
                 TEvReplFinished::TInfoPtr replInfo,
                 TBlobIdQueuePtr blobsToReplicatePtr,
-                TBlobIdQueuePtr unreplicatedBlobsPtr)
+                TBlobIdQueuePtr unreplicatedBlobsPtr,
+                TMilestoneQueue milestoneQueue)
             : ReplCtx(std::move(replCtx))
             , GInfo(std::move(ginfo))
             , StartKey(startKey)
             , ReplInfo(replInfo)
             , BlobsToReplicatePtr(std::move(blobsToReplicatePtr))
             , UnreplicatedBlobsPtr(std::move(unreplicatedBlobsPtr))
+            , MilestoneQueue(std::move(milestoneQueue))
         {}
     };
 
@@ -265,6 +310,7 @@ namespace NKikimr {
         TBlobIdQueuePtr BlobsToReplicatePtr;
         TBlobIdQueuePtr UnreplicatedBlobsPtr;
         TUnreplicatedBlobRecords UnreplicatedBlobRecords;
+        TMilestoneQueue MilestoneQueue;
         std::optional<std::pair<TVDiskID, TActorId>> Donor;
 
         // parameters from planner
@@ -305,7 +351,7 @@ namespace NKikimr {
             for (const auto& proxy : DiskProxySet) {
                 dropDonor = dropDonor && proxy && proxy->NoTransientErrors();
             }
-            ReplInfo->Finish(LastKey, Eof, Donor && dropDonor, std::move(UnreplicatedBlobRecords));
+            ReplInfo->Finish(LastKey, Eof, Donor && dropDonor, std::move(UnreplicatedBlobRecords), std::move(MilestoneQueue));
 
             TProxyStat stat;
             for (const TVDiskProxyPtr& p : DiskProxySet) {
@@ -326,7 +372,8 @@ namespace NKikimr {
             STLOG(PRI_DEBUG, BS_REPL, BSVR02, VDISKP(ReplCtx->VCtx->VDiskLogPrefix, "THullReplJobActor::Bootstrap"));
 
             TimeAccount.SetState(ETimeState::PREPARE_PLAN);
-            auto actor = std::make_unique<THullReplPlannerActor>(ReplCtx, GInfo, StartKey, ReplInfo, BlobsToReplicatePtr, UnreplicatedBlobsPtr);
+            auto actor = std::make_unique<THullReplPlannerActor>(ReplCtx, GInfo, StartKey, ReplInfo, BlobsToReplicatePtr,
+                UnreplicatedBlobsPtr, std::move(MilestoneQueue));
             auto aid = RunInBatchPool(ActorContext(), actor.release());
             ActiveActors.Insert(aid, __FILE__, __LINE__, TActivationContext::AsActorContext(), NKikimrServices::BLOBSTORAGE);
             Become(&TThis::StatePreparePlan);
@@ -338,6 +385,7 @@ namespace NKikimr {
             RecoveryMachine = std::move(ev->Get()->RecoveryMachine);
             LastKey = ev->Get()->LastKey;
             Eof = ev->Get()->Eof;
+            MilestoneQueue = std::move(ev->Get()->MilestoneQueue);
 
             for (const TLogoBlobID& id : ev->Get()->DroppedBlobs) {
                 DropUnreplicatedBlobRecord(id);
@@ -940,7 +988,8 @@ namespace NKikimr {
                 TBlobIdQueuePtr&& blobsToReplicatePtr,
                 TBlobIdQueuePtr&& unreplicatedBlobsPtr,
                 const std::optional<std::pair<TVDiskID, TActorId>>& donor,
-                TUnreplicatedBlobRecords&& ubr)
+                TUnreplicatedBlobRecords&& ubr,
+                TMilestoneQueue&& milestoneQueue)
             : TActorBootstrapped<THullReplJobActor>()
             , ReplCtx(std::move(replCtx))
             , GInfo(ReplCtx->GInfo) // it is safe to take it here
@@ -956,6 +1005,7 @@ namespace NKikimr {
             , BlobsToReplicatePtr(std::move(blobsToReplicatePtr))
             , UnreplicatedBlobsPtr(std::move(unreplicatedBlobsPtr))
             , UnreplicatedBlobRecords(std::move(ubr))
+            , MilestoneQueue(std::move(milestoneQueue))
             , Donor(donor)
         {
             if (Donor) {
@@ -987,10 +1037,12 @@ namespace NKikimr {
             TBlobIdQueuePtr blobsToReplicatePtr,
             TBlobIdQueuePtr unreplicatedBlobsPtr,
             const std::optional<std::pair<TVDiskID, TActorId>>& donor,
-            TUnreplicatedBlobRecords&& ubr)
+            TUnreplicatedBlobRecords&& ubr,
+            TMilestoneQueue&& milestoneQueue)
     {
         return new THullReplJobActor(std::move(replCtx), parentId, startKey, std::move(queueActorMapPtr),
-            std::move(blobsToReplicatePtr), std::move(unreplicatedBlobsPtr), donor, std::move(ubr));
+            std::move(blobsToReplicatePtr), std::move(unreplicatedBlobsPtr), donor, std::move(ubr),
+            std::move(milestoneQueue));
     }
 
 } // NKikimr
diff --git a/ydb/core/blobstorage/vdisk/repl/blobstorage_hullrepljob.h b/ydb/core/blobstorage/vdisk/repl/blobstorage_hullrepljob.h
@@ -18,6 +18,7 @@ namespace NKikimr {
             TBlobIdQueuePtr blobsToReplicatePtr,
             TBlobIdQueuePtr unreplicatedBlobsPtr,
             const std::optional<std::pair<TVDiskID, TActorId>>& donor,
-            TUnreplicatedBlobRecords&& ubr);
+            TUnreplicatedBlobRecords&& ubr,
+            TMilestoneQueue&& milestoneQueue);
 
 } // NKikimr
diff --git a/ydb/core/blobstorage/vdisk/repl/blobstorage_repl.cpp b/ydb/core/blobstorage/vdisk/repl/blobstorage_repl.cpp
@@ -167,6 +167,7 @@ namespace NKikimr {
         TBlobIdQueuePtr BlobsToReplicatePtr;
         TBlobIdQueuePtr UnreplicatedBlobsPtr = std::make_shared<TBlobIdQueue>();
         TUnreplicatedBlobRecords UnreplicatedBlobRecords;
+        TMilestoneQueue MilestoneQueue;
         TActorId ReplJobActorId;
         std::list<std::optional<TDonorQueueItem>> DonorQueue;
         std::deque<std::pair<TVDiskID, TActorId>> Donors;
@@ -356,6 +357,7 @@ namespace NKikimr {
 
             UnrecoveredNonphantomBlobs |= info->UnrecoveredNonphantomBlobs;
             UnreplicatedBlobRecords = std::move(info->UnreplicatedBlobRecords);
+            MilestoneQueue = std::move(info->MilestoneQueue);
 
             if (info->ItemsRecovered > 0) {
                 ResetReplProgressTimer(false);
@@ -453,7 +455,8 @@ namespace NKikimr {
                 donor->NodeId << ":" << donor->PDiskId << ":" << donor->VSlotId << "}") : "generic"));
             ReplJobActorId = Register(CreateReplJobActor(ReplCtx, SelfId(), from, QueueActorMapPtr,
                 BlobsToReplicatePtr, UnreplicatedBlobsPtr, donor ? std::make_optional(std::make_pair(
-                donor->VDiskId, donor->QueueActorId)) : std::nullopt, std::move(UnreplicatedBlobRecords)));
+                donor->VDiskId, donor->QueueActorId)) : std::nullopt, std::move(UnreplicatedBlobRecords),
+                std::move(MilestoneQueue)));
         }
 
         template<typename Iter>
diff --git a/ydb/core/blobstorage/vdisk/repl/blobstorage_repl.h b/ydb/core/blobstorage/vdisk/repl/blobstorage_repl.h
@@ -48,6 +48,49 @@ namespace NKikimr {
         }
     };
 
+    struct TMilestoneQueue {
+        bool Valid = false;
+        std::deque<std::tuple<TLogoBlobID, ui64, ui64>> Items; // id, items, units
+
+        void PopIfNeeded(const TLogoBlobID& id) {
+            while (Valid && !Items.empty() && std::get<0>(Items.front()) <= id) {
+                Items.pop_front();
+            }
+        }
+
+        bool Match(const TLogoBlobID& id, ui64 *totalItems, ui64 *totalUnits) {
+            if (Valid && !Items.empty() && std::get<0>(Items.front()) <= id) {
+                *totalItems += std::get<1>(Items.front());
+                *totalUnits += std::get<2>(Items.front());
+                return true;
+            }
+            return false;
+        }
+
+        void Push(const TLogoBlobID& id, ui64 units) {
+            if (!Valid) {
+                Y_DEBUG_ABORT_UNLESS(Items.empty() || std::get<0>(Items.back()) < id);
+                if (Items.empty() || std::get<1>(Items.back()) == 1000) {
+                    Items.emplace_back(id, 0, 0);
+                }
+                ++std::get<1>(Items.back());
+                std::get<2>(Items.back()) += units;
+            }
+        }
+
+        void Finish() {
+            if (!std::exchange(Valid, true)) {
+                ui64 totalItems = 0;
+                ui64 totalUnits = 0;
+                for (auto it = Items.rbegin(); it != Items.rend(); ++it) {
+                    auto& [id, items, units] = *it;
+                    items = totalItems += items;
+                    units = totalUnits += units;
+                }
+            }
+        }
+    };
+
     using TBlobIdQueuePtr = std::shared_ptr<TBlobIdQueue>;
 
     struct TUnreplicatedBlobRecord { // for monitoring purposes
@@ -122,13 +165,16 @@ namespace NKikimr {
             std::unique_ptr<NRepl::TProxyStat> ProxyStat;
 
             TUnreplicatedBlobRecords UnreplicatedBlobRecords;
+            TMilestoneQueue MilestoneQueue;
 
-            void Finish(const TLogoBlobID &keyPos, bool eof, bool dropDonor, TUnreplicatedBlobRecords&& ubr) {
+            void Finish(const TLogoBlobID &keyPos, bool eof, bool dropDonor, TUnreplicatedBlobRecords&& ubr,
+                    TMilestoneQueue&& milestoneQueue) {
                 End = TAppData::TimeProvider->Now();
                 KeyPos = keyPos;
                 Eof = eof;
                 DropDonor = dropDonor;
                 UnreplicatedBlobRecords = std::move(ubr);
+                MilestoneQueue = std::move(milestoneQueue);
             }
 
             TString ToString() const;
