Skip to content

Commit 0201d66

Browse files
alexvrualexvru
authored
Added milestone queue in replication to reduce work unit computations (#6561)
1 parent ee02567 commit 0201d66

File tree

4 files changed

+134
-32
lines changed

4 files changed

+134
-32
lines changed

‎ydb/core/blobstorage/vdisk/repl/blobstorage_hullrepljob.cpp

Lines changed: 81 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -26,13 +26,15 @@ namespace NKikimr {
2626
TLogoBlobID LastKey;
2727
bool Eof;
2828
std::deque<TLogoBlobID> DroppedBlobs;
29+
TMilestoneQueue MilestoneQueue;
2930

3031
TEvReplPlanFinished(std::unique_ptr<TRecoveryMachine>&& recoveryMachine, const TLogoBlobID& lastKey, bool eof,
31-
std::deque<TLogoBlobID>&& droppedBlobs)
32+
std::deque<TLogoBlobID>&& droppedBlobs, TMilestoneQueue&& milestoneQueue)
3233
: RecoveryMachine(std::move(recoveryMachine))
3334
, LastKey(lastKey)
3435
, Eof(eof)
3536
, DroppedBlobs(std::move(droppedBlobs))
37+
, MilestoneQueue(std::move(milestoneQueue))
3638
{}
3739
};
3840

@@ -52,6 +54,7 @@ namespace NKikimr {
5254
std::deque<TLogoBlobID> DroppedBlobs;
5355
ui64 QuantumBytes = 0;
5456
bool AddingTasks = true;
57+
TMilestoneQueue MilestoneQueue;
5558

5659
public:
5760
void Bootstrap(const TActorId& parentId) {
@@ -106,43 +109,83 @@ namespace NKikimr {
106109
} else {
107110
// scan through the index until we have enough blobs to recover or the time is out
108111
const TBlobStorageGroupInfo::TTopology& topology = *ReplCtx->VCtx->Top;
109-
for (it.Seek(StartKey); it.Valid(); it.Next()) {
110-
StartKey = it.GetCurKey().LogoBlobID();
112+
113+
if (StartKey == TLogoBlobID()) {
114+
it.SeekToFirst();
115+
} else {
116+
it.Seek(StartKey);
117+
if (it.Valid() && it.GetCurKey() == StartKey) {
118+
it.Next();
119+
}
120+
}
121+
122+
auto checkRestart = [&] {
111123
if (++counter % 1024 == 0 && GetCycleCountFast() >= plannedEndTime) {
112124
// we have event processing timer expired, restart processing later with new snapshot starting
113125
// with current key
114126
Send(ReplCtx->SkeletonId, new TEvTakeHullSnapshot(true));
115-
return;
116-
} else if (AddingTasks) {
127+
return true;
128+
}
129+
return false;
130+
};
131+
132+
if (AddingTasks) {
133+
for (; it.Valid(); it.Next()) {
134+
if (checkRestart()) {
135+
return;
136+
}
137+
138+
StartKey = it.GetCurKey().LogoBlobID();
139+
117140
// we still have some space in recovery machine logic, so we can add new item
118141
ProcessItem(it, *barriers, allowKeepFlags);
119-
} else {
120-
// no space in recovery machine logic, but we still have to count remaining work
121-
const TMemRecLogoBlob memRec = it.GetMemRec();
122-
const TIngress ingress = memRec.GetIngress();
123-
const auto parts = ingress.PartsWeMustHaveLocally(&topology, ReplCtx->VCtx->ShortSelfVDisk,
124-
StartKey) - ingress.LocalParts(topology.GType);
125-
if (!parts.Empty() && barriers->Keep(StartKey, memRec, {}, allowKeepFlags,
126-
true /*allowGarbageCollection*/).KeepData) {
127-
++ReplInfo->ItemsTotal;
128-
ReplInfo->WorkUnitsTotal += StartKey.BlobSize();
129-
}
142+
MilestoneQueue.PopIfNeeded(StartKey);
130143

131-
if (!KeyToResumeNextTime) {
132-
// this is first valid key that is not processed with ProcessItem, so we remember it to
133-
// start next quantum with this exact key
134-
KeyToResumeNextTime.emplace(StartKey);
144+
if (!AddingTasks) { // we have finished adding tasks after this key, remember it
145+
it.Next();
146+
Y_ABORT_UNLESS(!KeyToResumeNextTime);
147+
if (it.Valid()) {
148+
KeyToResumeNextTime.emplace(it.GetCurKey().LogoBlobID());
149+
}
150+
break;
135151
}
136152
}
153+
eof = !it.Valid(); // we finish this quantum when there are no more tasks to generate
137154
}
138155

139-
// we shall run next quantum only if we have KeyToResumeNextTime filled in
140-
eof = !KeyToResumeNextTime;
156+
for (; it.Valid(); it.Next()) {
157+
if (checkRestart()) {
158+
return;
159+
}
160+
161+
StartKey = it.GetCurKey().LogoBlobID();
162+
163+
// check the milestone queue, if we have requested blob
164+
if (MilestoneQueue.Match(StartKey, &ReplInfo->ItemsTotal, &ReplInfo->WorkUnitsTotal)) {
165+
break;
166+
}
167+
168+
// no space in recovery machine logic, but we still have to count remaining work
169+
const TMemRecLogoBlob memRec = it.GetMemRec();
170+
const TIngress ingress = memRec.GetIngress();
171+
const auto parts = ingress.PartsWeMustHaveLocally(&topology, ReplCtx->VCtx->ShortSelfVDisk,
172+
StartKey) - ingress.LocalParts(topology.GType);
173+
if (!parts.Empty() && barriers->Keep(StartKey, memRec, {}, allowKeepFlags,
174+
true /*allowGarbageCollection*/).KeepData) {
175+
++ReplInfo->ItemsTotal;
176+
ReplInfo->WorkUnitsTotal += StartKey.BlobSize();
177+
MilestoneQueue.Push(StartKey, StartKey.BlobSize());
178+
}
179+
}
180+
181+
if (!it.Valid()) {
182+
MilestoneQueue.Finish();
183+
}
141184
}
142185

143186
// the planning stage has finished, issue reply to the job actor
144187
Send(Recipient, new TEvReplPlanFinished(std::move(RecoveryMachine), KeyToResumeNextTime.value_or(TLogoBlobID()),
145-
eof, std::move(DroppedBlobs)));
188+
eof, std::move(DroppedBlobs), std::move(MilestoneQueue)));
146189

147190
// finish processing for this actor
148191
PassAway();
@@ -219,13 +262,15 @@ namespace NKikimr {
219262
const TLogoBlobID &startKey,
220263
TEvReplFinished::TInfoPtr replInfo,
221264
TBlobIdQueuePtr blobsToReplicatePtr,
222-
TBlobIdQueuePtr unreplicatedBlobsPtr)
265+
TBlobIdQueuePtr unreplicatedBlobsPtr,
266+
TMilestoneQueue milestoneQueue)
223267
: ReplCtx(std::move(replCtx))
224268
, GInfo(std::move(ginfo))
225269
, StartKey(startKey)
226270
, ReplInfo(replInfo)
227271
, BlobsToReplicatePtr(std::move(blobsToReplicatePtr))
228272
, UnreplicatedBlobsPtr(std::move(unreplicatedBlobsPtr))
273+
, MilestoneQueue(std::move(milestoneQueue))
229274
{}
230275
};
231276

@@ -265,6 +310,7 @@ namespace NKikimr {
265310
TBlobIdQueuePtr BlobsToReplicatePtr;
266311
TBlobIdQueuePtr UnreplicatedBlobsPtr;
267312
TUnreplicatedBlobRecords UnreplicatedBlobRecords;
313+
TMilestoneQueue MilestoneQueue;
268314
std::optional<std::pair<TVDiskID, TActorId>> Donor;
269315

270316
// parameters from planner
@@ -305,7 +351,7 @@ namespace NKikimr {
305351
for (const auto& proxy : DiskProxySet) {
306352
dropDonor = dropDonor && proxy && proxy->NoTransientErrors();
307353
}
308-
ReplInfo->Finish(LastKey, Eof, Donor && dropDonor, std::move(UnreplicatedBlobRecords));
354+
ReplInfo->Finish(LastKey, Eof, Donor && dropDonor, std::move(UnreplicatedBlobRecords), std::move(MilestoneQueue));
309355

310356
TProxyStat stat;
311357
for (const TVDiskProxyPtr& p : DiskProxySet) {
@@ -326,7 +372,8 @@ namespace NKikimr {
326372
STLOG(PRI_DEBUG, BS_REPL, BSVR02, VDISKP(ReplCtx->VCtx->VDiskLogPrefix, "THullReplJobActor::Bootstrap"));
327373

328374
TimeAccount.SetState(ETimeState::PREPARE_PLAN);
329-
auto actor = std::make_unique<THullReplPlannerActor>(ReplCtx, GInfo, StartKey, ReplInfo, BlobsToReplicatePtr, UnreplicatedBlobsPtr);
375+
auto actor = std::make_unique<THullReplPlannerActor>(ReplCtx, GInfo, StartKey, ReplInfo, BlobsToReplicatePtr,
376+
UnreplicatedBlobsPtr, std::move(MilestoneQueue));
330377
auto aid = RunInBatchPool(ActorContext(), actor.release());
331378
ActiveActors.Insert(aid, __FILE__, __LINE__, TActivationContext::AsActorContext(), NKikimrServices::BLOBSTORAGE);
332379
Become(&TThis::StatePreparePlan);
@@ -338,6 +385,7 @@ namespace NKikimr {
338385
RecoveryMachine = std::move(ev->Get()->RecoveryMachine);
339386
LastKey = ev->Get()->LastKey;
340387
Eof = ev->Get()->Eof;
388+
MilestoneQueue = std::move(ev->Get()->MilestoneQueue);
341389

342390
for (const TLogoBlobID& id : ev->Get()->DroppedBlobs) {
343391
DropUnreplicatedBlobRecord(id);
@@ -940,7 +988,8 @@ namespace NKikimr {
940988
TBlobIdQueuePtr&& blobsToReplicatePtr,
941989
TBlobIdQueuePtr&& unreplicatedBlobsPtr,
942990
const std::optional<std::pair<TVDiskID, TActorId>>& donor,
943-
TUnreplicatedBlobRecords&& ubr)
991+
TUnreplicatedBlobRecords&& ubr,
992+
TMilestoneQueue&& milestoneQueue)
944993
: TActorBootstrapped<THullReplJobActor>()
945994
, ReplCtx(std::move(replCtx))
946995
, GInfo(ReplCtx->GInfo) // it is safe to take it here
@@ -956,6 +1005,7 @@ namespace NKikimr {
9561005
, BlobsToReplicatePtr(std::move(blobsToReplicatePtr))
9571006
, UnreplicatedBlobsPtr(std::move(unreplicatedBlobsPtr))
9581007
, UnreplicatedBlobRecords(std::move(ubr))
1008+
, MilestoneQueue(std::move(milestoneQueue))
9591009
, Donor(donor)
9601010
{
9611011
if (Donor) {
@@ -987,10 +1037,12 @@ namespace NKikimr {
9871037
TBlobIdQueuePtr blobsToReplicatePtr,
9881038
TBlobIdQueuePtr unreplicatedBlobsPtr,
9891039
const std::optional<std::pair<TVDiskID, TActorId>>& donor,
990-
TUnreplicatedBlobRecords&& ubr)
1040+
TUnreplicatedBlobRecords&& ubr,
1041+
TMilestoneQueue&& milestoneQueue)
9911042
{
9921043
return new THullReplJobActor(std::move(replCtx), parentId, startKey, std::move(queueActorMapPtr),
993-
std::move(blobsToReplicatePtr), std::move(unreplicatedBlobsPtr), donor, std::move(ubr));
1044+
std::move(blobsToReplicatePtr), std::move(unreplicatedBlobsPtr), donor, std::move(ubr),
1045+
std::move(milestoneQueue));
9941046
}
9951047

9961048
} // NKikimr

‎ydb/core/blobstorage/vdisk/repl/blobstorage_hullrepljob.h

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ namespace NKikimr {
1818
TBlobIdQueuePtr blobsToReplicatePtr,
1919
TBlobIdQueuePtr unreplicatedBlobsPtr,
2020
const std::optional<std::pair<TVDiskID, TActorId>>& donor,
21-
TUnreplicatedBlobRecords&& ubr);
21+
TUnreplicatedBlobRecords&& ubr,
22+
TMilestoneQueue&& milestoneQueue);
2223

2324
} // NKikimr

‎ydb/core/blobstorage/vdisk/repl/blobstorage_repl.cpp

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -167,6 +167,7 @@ namespace NKikimr {
167167
TBlobIdQueuePtr BlobsToReplicatePtr;
168168
TBlobIdQueuePtr UnreplicatedBlobsPtr = std::make_shared<TBlobIdQueue>();
169169
TUnreplicatedBlobRecords UnreplicatedBlobRecords;
170+
TMilestoneQueue MilestoneQueue;
170171
TActorId ReplJobActorId;
171172
std::list<std::optional<TDonorQueueItem>> DonorQueue;
172173
std::deque<std::pair<TVDiskID, TActorId>> Donors;
@@ -356,6 +357,7 @@ namespace NKikimr {
356357

357358
UnrecoveredNonphantomBlobs |= info->UnrecoveredNonphantomBlobs;
358359
UnreplicatedBlobRecords = std::move(info->UnreplicatedBlobRecords);
360+
MilestoneQueue = std::move(info->MilestoneQueue);
359361

360362
if (info->ItemsRecovered > 0) {
361363
ResetReplProgressTimer(false);
@@ -453,7 +455,8 @@ namespace NKikimr {
453455
donor->NodeId << ":" << donor->PDiskId << ":" << donor->VSlotId << "}") : "generic"));
454456
ReplJobActorId = Register(CreateReplJobActor(ReplCtx, SelfId(), from, QueueActorMapPtr,
455457
BlobsToReplicatePtr, UnreplicatedBlobsPtr, donor ? std::make_optional(std::make_pair(
456-
donor->VDiskId, donor->QueueActorId)) : std::nullopt, std::move(UnreplicatedBlobRecords)));
458+
donor->VDiskId, donor->QueueActorId)) : std::nullopt, std::move(UnreplicatedBlobRecords),
459+
std::move(MilestoneQueue)));
457460
}
458461

459462
template<typename Iter>

‎ydb/core/blobstorage/vdisk/repl/blobstorage_repl.h

Lines changed: 47 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,49 @@ namespace NKikimr {
4848
}
4949
};
5050

51+
struct TMilestoneQueue {
52+
bool Valid = false;
53+
std::deque<std::tuple<TLogoBlobID, ui64, ui64>> Items; // id, items, units
54+
55+
void PopIfNeeded(const TLogoBlobID& id) {
56+
while (Valid && !Items.empty() && std::get<0>(Items.front()) <= id) {
57+
Items.pop_front();
58+
}
59+
}
60+
61+
bool Match(const TLogoBlobID& id, ui64 *totalItems, ui64 *totalUnits) {
62+
if (Valid && !Items.empty() && std::get<0>(Items.front()) <= id) {
63+
*totalItems += std::get<1>(Items.front());
64+
*totalUnits += std::get<2>(Items.front());
65+
return true;
66+
}
67+
return false;
68+
}
69+
70+
void Push(const TLogoBlobID& id, ui64 units) {
71+
if (!Valid) {
72+
Y_DEBUG_ABORT_UNLESS(Items.empty() || std::get<0>(Items.back()) < id);
73+
if (Items.empty() || std::get<1>(Items.back()) == 1000) {
74+
Items.emplace_back(id, 0, 0);
75+
}
76+
++std::get<1>(Items.back());
77+
std::get<2>(Items.back()) += units;
78+
}
79+
}
80+
81+
void Finish() {
82+
if (!std::exchange(Valid, true)) {
83+
ui64 totalItems = 0;
84+
ui64 totalUnits = 0;
85+
for (auto it = Items.rbegin(); it != Items.rend(); ++it) {
86+
auto& [id, items, units] = *it;
87+
items = totalItems += items;
88+
units = totalUnits += units;
89+
}
90+
}
91+
}
92+
};
93+
5194
using TBlobIdQueuePtr = std::shared_ptr<TBlobIdQueue>;
5295

5396
struct TUnreplicatedBlobRecord { // for monitoring purposes
@@ -122,13 +165,16 @@ namespace NKikimr {
122165
std::unique_ptr<NRepl::TProxyStat> ProxyStat;
123166

124167
TUnreplicatedBlobRecords UnreplicatedBlobRecords;
168+
TMilestoneQueue MilestoneQueue;
125169

126-
void Finish(const TLogoBlobID &keyPos, bool eof, bool dropDonor, TUnreplicatedBlobRecords&& ubr) {
170+
void Finish(const TLogoBlobID &keyPos, bool eof, bool dropDonor, TUnreplicatedBlobRecords&& ubr,
171+
TMilestoneQueue&& milestoneQueue) {
127172
End = TAppData::TimeProvider->Now();
128173
KeyPos = keyPos;
129174
Eof = eof;
130175
DropDonor = dropDonor;
131176
UnreplicatedBlobRecords = std::move(ubr);
177+
MilestoneQueue = std::move(milestoneQueue);
132178
}
133179

134180
TString ToString() const;

0 commit comments

Comments
 (0)