Skip to content

Commit 745a010

Browse files
alexvrualexvru
authored
Finish implementing shred logic in VDisk (#14534)
1 parent 7a0b0c7 commit 745a010

24 files changed

+430
-223
lines changed

‎ydb/core/base/blobstorage.h

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -752,6 +752,10 @@ struct TEvBlobStorage {
752752
EvHugeShredNotify,
753753
EvHugeShredNotifyResult,
754754
EvNotifyChunksDeleted,
755+
EvListChunks,
756+
EvListChunksResult,
757+
EvHugeQueryForbiddenChunks,
758+
EvHugeForbiddenChunks,
755759

756760
EvYardInitResult = EvPut + 9 * 512, /// 268 636 672
757761
EvLogResult,

‎ydb/core/blobstorage/nodewarden/node_warden_impl.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -770,7 +770,7 @@ void TNodeWarden::Handle(TEvBlobStorage::TEvControllerNodeServiceSetUpdate::TPtr
770770
auto issueShredRequestToPDisk = [&] {
771771
const ui64 cookie = ++LastShredCookie;
772772
ShredInFlight.emplace(cookie, key);
773-
pdisk.ShredCookies.insert(cookie);
773+
pdisk.ShredCookies.emplace(cookie, generation);
774774

775775
const TActorId actorId = SelfId();
776776
auto ev = std::make_unique<NPDisk::TEvShredPDisk>(generation);

‎ydb/core/blobstorage/nodewarden/node_warden_impl.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@ namespace NKikimr::NStorage {
8787

8888
std::optional<ui64> ShredGenerationIssued;
8989
std::variant<std::monostate, ui64, TString> ShredState; // not issued, finished with generation, aborted
90-
THashSet<ui64> ShredCookies;
90+
THashMap<ui64, ui64> ShredCookies;
9191

9292
TPDiskRecord(NKikimrBlobStorage::TNodeWardenServiceSet::TPDisk record)
9393
: Record(std::move(record))

‎ydb/core/blobstorage/nodewarden/node_warden_pdisk.cpp

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -224,8 +224,10 @@ namespace NKikimr::NStorage {
224224
Y_ABORT_UNLESS(jt != PDiskByPath.end() && jt->second.RunningPDiskId == it->first);
225225
pending = std::move(jt->second.Pending);
226226
PDiskByPath.erase(jt);
227-
for (ui64 cookie : it->second.ShredCookies) {
228-
ShredInFlight.erase(cookie);
227+
auto& cookies = it->second.ShredCookies;
228+
for (auto it = cookies.begin(); it != cookies.end(); ) {
229+
const auto& [cookie, generation] = *it++;
230+
ProcessShredStatus(cookie, generation, "pdisk has been restarted");
229231
}
230232
LocalPDisks.erase(it);
231233
PDiskRestartInFlight.erase(pdiskId);

‎ydb/core/blobstorage/pdisk/mock/pdisk_mock.cpp

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -989,6 +989,8 @@ class TPDiskMockActor : public TActorBootstrapped<TPDiskMockActor> {
989989
Become(&TThis::StateNormal);
990990
}
991991

992+
void Ignore() {}
993+
992994
STRICT_STFUNC(StateNormal,
993995
hFunc(NPDisk::TEvYardInit, Handle);
994996
hFunc(NPDisk::TEvLog, Handle);
@@ -1010,6 +1012,8 @@ class TPDiskMockActor : public TActorBootstrapped<TPDiskMockActor> {
10101012
hFunc(NPDisk::TEvWriteMetadata, Handle);
10111013

10121014
cFunc(EvBecomeError, HandleMoveToErrorState);
1015+
1016+
cFunc(TEvBlobStorage::EvMarkDirty, Ignore);
10131017
)
10141018

10151019
STRICT_STFUNC(StateError,
@@ -1030,6 +1034,8 @@ class TPDiskMockActor : public TActorBootstrapped<TPDiskMockActor> {
10301034

10311035
cFunc(TEvents::TSystem::Wakeup, ReportMetrics);
10321036
cFunc(EvBecomeNormal, HandleMoveToNormalState);
1037+
1038+
cFunc(TEvBlobStorage::EvMarkDirty, Ignore);
10331039
)
10341040
};
10351041

‎ydb/core/blobstorage/ut_vdisk/lib/test_huge.cpp

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -293,6 +293,7 @@ class THugeModuleTestActor : public TActorBootstrapped<THugeModuleTestActor> {
293293

294294
STRICT_STFUNC(StateWorking,
295295
HFunc(TEvHullLogHugeBlob, Handle);
296+
cFunc(TEvBlobStorage::EvNotifyChunksDeleted, []{});
296297
)
297298

298299
public:

‎ydb/core/blobstorage/vdisk/common/vdisk_private_events.h

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -156,8 +156,8 @@ namespace NKikimr {
156156
};
157157

158158
struct TEvHullShredDefrag : TEventLocal<TEvHullShredDefrag, TEvBlobStorage::EvHullShredDefrag> {
159-
std::vector<TChunkIdx> ChunksToShred;
160-
TEvHullShredDefrag(std::vector<TChunkIdx> chunksToShred) : ChunksToShred(std::move(chunksToShred)) {}
159+
THashSet<TChunkIdx> ChunksToShred;
160+
TEvHullShredDefrag(THashSet<TChunkIdx> chunksToShred) : ChunksToShred(std::move(chunksToShred)) {}
161161
};
162162

163163
struct TEvHullShredDefragResult : TEventLocal<TEvHullShredDefragResult, TEvBlobStorage::EvHullShredDefragResult> {
@@ -171,11 +171,16 @@ namespace NKikimr {
171171
: Lsn(lsn)
172172
, Chunks(std::move(chunks))
173173
{}
174+
};
174175

175-
TEvNotifyChunksDeleted(const TEvNotifyChunksDeleted& x)
176-
: Lsn(x.Lsn)
177-
, Chunks(x.Chunks)
178-
{}
176+
struct TEvListChunks : TEventLocal<TEvListChunks, TEvBlobStorage::EvListChunks> {
177+
THashSet<TChunkIdx> ChunksOfInterest;
178+
TEvListChunks(THashSet<TChunkIdx> chunksOfInterest) : ChunksOfInterest(std::move(chunksOfInterest)) {}
179+
};
180+
181+
struct TEvListChunksResult : TEventLocal<TEvListChunksResult, TEvBlobStorage::EvListChunksResult> {
182+
THashSet<TChunkIdx> ChunksHuge;
183+
THashSet<TChunkIdx> ChunksSyncLog;
179184
};
180185

181186
} // NKikimr

‎ydb/core/blobstorage/vdisk/defrag/defrag_actor.cpp

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -331,6 +331,16 @@ namespace NKikimr {
331331
RunDefragIfAny(ctx);
332332
}
333333

334+
void Handle(TEvNotifyChunksDeleted::TPtr ev, const TActorContext& /*ctx*/) {
335+
for (TTask& task : WaitQueue) {
336+
if (auto *ptr = std::get_if<TEvHullShredDefrag::TPtr>(&task.Request)) {
337+
for (const TChunkIdx chunkId : ev->Get()->Chunks) {
338+
(*ptr)->Get()->ChunksToShred.erase(chunkId);
339+
}
340+
}
341+
}
342+
}
343+
334344
void Die(const TActorContext& ctx) override {
335345
ActiveActors.KillAndClear(ctx);
336346
TActorBootstrapped::Die(ctx);
@@ -447,6 +457,7 @@ namespace NKikimr {
447457
HFunc(TEvSublogLine, Handle)
448458
HFunc(TEvDefragStartQuantum, Handle)
449459
HFunc(TEvDefragQuantumResult, Handle)
460+
HFunc(TEvNotifyChunksDeleted, Handle)
450461
);
451462

452463
public:

‎ydb/core/blobstorage/vdisk/defrag/defrag_quantum.cpp

Lines changed: 30 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -76,11 +76,9 @@ namespace NKikimr {
7676
Yield();
7777
}
7878
ChunksToDefrag.emplace(findChunks.GetChunksToDefrag(DCtx->MaxChunksToDefrag));
79-
} else {
80-
Y_ABORT_UNLESS(*ChunksToDefrag || ChunksToDefrag->IsShred());
8179
}
82-
if (*ChunksToDefrag || ChunksToDefrag->IsShred()) {
83-
const bool isShred = ChunksToDefrag->IsShred();
80+
if (*ChunksToDefrag || ChunksToDefrag->IsShred) {
81+
const bool isShred = ChunksToDefrag->IsShred;
8482

8583
TDefragChunks lockedChunks;
8684

@@ -97,8 +95,27 @@ namespace NKikimr {
9795
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD11, DCtx->VCtx->VDiskLogPrefix << "locked chunks",
9896
(ActorId, SelfActorId), (LockedChunks, lockedChunks));
9997
} else {
98+
auto forbiddenChunks = GetForbiddenChunks();
99+
100100
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD14, DCtx->VCtx->VDiskLogPrefix
101-
<< "commencing shredding", (ActorId, SelfActorId), (ChunksToShred, ChunksToDefrag->ChunksToShred));
101+
<< "commencing shredding", (ActorId, SelfActorId), (ChunksToShred, ChunksToDefrag->ChunksToShred),
102+
(ForbiddenChunks, forbiddenChunks));
103+
104+
// filter chunks to shred via forbidden chunks
105+
auto& chunksToShred = ChunksToDefrag->ChunksToShred;
106+
for (const TChunkIdx chunkIdx : std::exchange(chunksToShred, {})) {
107+
if (forbiddenChunks.contains(chunkIdx)) {
108+
chunksToShred.insert(chunkIdx);
109+
}
110+
}
111+
112+
// check if we have something remaining to process
113+
if (chunksToShred.empty()) {
114+
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD15, DCtx->VCtx->VDiskLogPrefix << "nothing to do",
115+
(ActorId, SelfActorId), (Stat, stat));
116+
Send(ParentActorId, new TEvDefragQuantumResult(std::move(stat)));
117+
return;
118+
}
102119
}
103120

104121
TDefragQuantumFindRecords findRecords(std::move(*ChunksToDefrag), lockedChunks);
@@ -126,6 +143,7 @@ namespace NKikimr {
126143
auto pred = [&](const auto& record) { return !set.contains(record.OldDiskPart.ChunkIdx); };
127144
auto range = std::ranges::remove_if(records, pred);
128145
records.erase(range.begin(), range.end());
146+
findRecords.SetLockedChunks(std::move(set));
129147
}
130148

131149
auto getSortedChunks = [&] {
@@ -198,6 +216,13 @@ namespace NKikimr {
198216
return res->Get()->LockedChunks;
199217
}
200218

219+
THashSet<TChunkIdx> GetForbiddenChunks() {
220+
TActivationContext::Send(new IEventHandle(TEvBlobStorage::EvHugeQueryForbiddenChunks, 0, DCtx->HugeKeeperId,
221+
SelfActorId, nullptr, 0));
222+
auto res = WaitForSpecificEvent<TEvHugeForbiddenChunks>(&TDefragQuantum::ProcessUnexpectedEvent);
223+
return res->Get()->ForbiddenChunks;
224+
}
225+
201226
void Compact(THashSet<ui64> tablesToCompact, bool needsFreshCompaction) {
202227
if (tablesToCompact) {
203228
Send(DCtx->SkeletonId, TEvCompactVDisk::Create(EHullDbType::LogoBlobs, std::move(tablesToCompact)));

‎ydb/core/blobstorage/vdisk/defrag/defrag_search.h

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -18,12 +18,14 @@ namespace NKikimr {
1818
TDefragChunks Chunks;
1919
ui32 FoundChunksToDefrag = 0;
2020
ui64 EstimatedSlotsCount = 0;
21+
bool IsShred = false;
2122
THashSet<TChunkIdx> ChunksToShred;
2223

23-
static TChunksToDefrag Shred(const auto& chunks) {
24-
TChunksToDefrag res;
25-
res.ChunksToShred = {chunks.begin(), chunks.end()};
26-
return res;
24+
static TChunksToDefrag Shred(THashSet<TChunkIdx> chunksToShred) {
25+
return {
26+
.IsShred = true,
27+
.ChunksToShred = std::move(chunksToShred),
28+
};
2729
}
2830

2931
void Output(IOutputStream &str) const {
@@ -40,10 +42,6 @@ namespace NKikimr {
4042
explicit operator bool() const {
4143
return !Chunks.empty();
4244
}
43-
44-
bool IsShred() const {
45-
return !ChunksToShred.empty();
46-
}
4745
};
4846

4947
struct TDefragRecord {
@@ -371,7 +369,7 @@ namespace NKikimr {
371369

372370
public:
373371
TDefragQuantumFindRecords(TChunksToDefrag&& chunksToDefrag, const TDefragChunks& locked) {
374-
if (chunksToDefrag.IsShred()) {
372+
if (chunksToDefrag.IsShred) {
375373
LockedChunks = Chunks = std::move(chunksToDefrag.ChunksToShred);
376374
} else {
377375
for (const auto& chunk : chunksToDefrag.Chunks) {
@@ -396,6 +394,10 @@ namespace NKikimr {
396394
NeedsFreshCompaction = false;
397395
}
398396

397+
void SetLockedChunks(THashSet<ui32> lockedChunks) {
398+
LockedChunks = std::move(lockedChunks);
399+
}
400+
399401
std::vector<TDefragRecord> GetRecordsToRewrite() { return std::move(RecsToRewrite); }
400402
THashSet<ui64> GetTablesToCompact() { return std::move(TablesToCompact); }
401403
bool GetNeedsFreshCompaction() const { return NeedsFreshCompaction; }

0 commit comments

Comments
 (0)