Skip to content

Commit d1b4cd3

Browse files
alexvrualexvru
authored
Defrag fixes for shredding (#14397)
1 parent 67a4afb commit d1b4cd3

23 files changed

+383
-382
lines changed

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -71,10 +71,12 @@ virtual void Scenario(const TActorContext &ctx) {
7171
SyncRunner->Run(ctx, CreateWaitForSync(SyncRunner->NotifyID(), Conf));
7272

7373
// check compaction result
74-
ui32 freedChunks = 0;
74+
THashSet<ui32> freedChunks;
7575
auto check = [&freedChunks] (TEvBlobStorage::TEvVDefragResult::TPtr &ev) {
7676
const auto &rec = ev->Get()->Record;
77-
freedChunks += rec.GetFreedChunks().size();
77+
for (const auto& chunk : rec.GetFreedChunks()) {
78+
freedChunks.insert(chunk);
79+
}
7880
STR << "FoundChunksToDefrag# " << rec.GetFoundChunksToDefrag()
7981
<< " RewrittenRecs# " << rec.GetRewrittenRecs()
8082
<< " RewrittenBytes# " << rec.GetRewrittenBytes()
@@ -98,7 +100,7 @@ virtual void Scenario(const TActorContext &ctx) {
98100
LOG_NOTICE(ctx, NActorsServices::TEST, " Defrag completed");
99101

100102
// check actually freed chunks
101-
UNIT_ASSERT_VALUES_EQUAL(freedChunks, 4);
103+
UNIT_ASSERT_VALUES_EQUAL(freedChunks.size(), 4);
102104
}
103105
SYNC_TEST_END(TDefrag50PercentGarbage, TSyncTestBase)
104106

‎ydb/core/blobstorage/vdisk/anubis_osiris/blobstorage_anubisfinder.cpp

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -88,8 +88,9 @@ namespace NKikimr {
8888
Y_UNUSED(subsMerger);
8989
// calculate keep status
9090
bool allowKeepFlags = HullCtx->AllowKeepFlags;
91-
NGc::TKeepStatus keep = brs->Keep(dbIt.GetCurKey(), dbMerger.GetMemRec(),
92-
{subsMerger, dbMerger}, allowKeepFlags, true /*allowGarbageCollection*/);
91+
NGc::TKeepStatus keep = brs->Keep(dbIt.GetCurKey(), dbMerger.GetMemRec(), {subsMerger.GetNumKeepFlags(),
92+
subsMerger.GetNumDoNotKeepFlags(), dbMerger.GetNumKeepFlags(), dbMerger.GetNumDoNotKeepFlags()},
93+
allowKeepFlags, true /*allowGarbageCollection*/);
9394
if (keep.KeepIndex && !keep.KeepByBarrier) {
9495
// we keep this record because of keep flags
9596
candidates.AddCandidate(dbIt.GetCurKey().LogoBlobID());

‎ydb/core/blobstorage/vdisk/balance/handoff_map.h

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ namespace NKikimr {
5959

6060
// Transforms record according to the built handoff map. It returns item we need to write, pointer is
6161
// valid until next call. Nullptr indicates that item is to be removed completely.
62-
void Transform(const TKey& /*key*/, TMemRec& /*memRec*/, TDataMerger& dataMerger, bool /*keepData*/) {
62+
void Transform(const TKey& /*key*/, TMemRec& /*memRec*/, TDataMerger& dataMerger) {
6363
// do nothing by default, all work is done in template specialization for logo blobs
6464
Counter++;
6565
Y_DEBUG_ABORT_UNLESS(dataMerger.Empty());
@@ -78,16 +78,9 @@ namespace NKikimr {
7878

7979
template<>
8080
inline void THandoffMap<TKeyLogoBlob, TMemRecLogoBlob>::Transform(const TKeyLogoBlob& key, TMemRecLogoBlob& memRec,
81-
TDataMerger& dataMerger, bool keepData) {
81+
TDataMerger& dataMerger) {
8282
Y_DEFER { Counter++; };
8383

84-
if (!keepData) {
85-
memRec.SetNoBlob();
86-
memRec.ClearLocalParts(Top->GType);
87-
dataMerger.MakeEmpty(); // collect all huge blobs in the merger
88-
return;
89-
}
90-
9184
if (!RunHandoff) {
9285
return;
9386
}

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

Lines changed: 4 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@ namespace NKikimr {
3131
, HugeKeeperId(hugeKeeperId)
3232
, DefragMonGroup(VCtx->VDiskCounters, "subsystem", "defrag")
3333
, RunDefragBySchedule(runDefrageBySchedule)
34-
, AddHeader(vconfig->AddHeader)
3534
, MaxChunksToDefrag(vconfig->MaxChunksToDefragInflight)
3635
{}
3736

@@ -236,7 +235,6 @@ namespace NKikimr {
236235
std::deque<TTask> WaitQueue;
237236
TActiveActors ActiveActors;
238237
TSublog<TCircleBufStringStream<81920>> Sublog = { true };
239-
ui32 MinHugeBlobInBytes;
240238

241239
friend class TActorBootstrapped<TDefragActor>;
242240

@@ -266,7 +264,7 @@ namespace NKikimr {
266264
++TotalDefragRuns;
267265
InProgress = true;
268266
ActiveActors.Insert(ctx.Register(CreateDefragQuantumActor(DCtx, GInfo->GetVDiskId(DCtx->VCtx->ShortSelfVDisk),
269-
std::visit(getChunksToDefrag, task.Request), MinHugeBlobInBytes)), __FILE__, __LINE__, ctx,
267+
std::visit(getChunksToDefrag, task.Request))), __FILE__, __LINE__, ctx,
270268
NKikimrServices::BLOBSTORAGE);
271269
}
272270

@@ -374,10 +372,6 @@ namespace NKikimr {
374372
ctx.Send(ev->Sender, new NMon::TEvHttpInfoRes(str.Str(), subrequest));
375373
}
376374

377-
void Handle(TEvMinHugeBlobSizeUpdate::TPtr ev) {
378-
MinHugeBlobInBytes = ev->Get()->MinHugeBlobInBytes;
379-
}
380-
381375
void RenderHtml(IOutputStream &str) const {
382376
HTML(str) {
383377
DIV_CLASS("panel panel-info") {
@@ -453,30 +447,26 @@ namespace NKikimr {
453447
HFunc(TEvSublogLine, Handle)
454448
HFunc(TEvDefragStartQuantum, Handle)
455449
HFunc(TEvDefragQuantumResult, Handle)
456-
hFunc(TEvMinHugeBlobSizeUpdate, Handle)
457450
);
458451

459452
public:
460453
static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
461454
return NKikimrServices::TActivity::BS_DEFRAG;
462455
}
463456

464-
TDefragActor(const std::shared_ptr<TDefragCtx> &dCtx, const TIntrusivePtr<TBlobStorageGroupInfo> &info,
465-
ui32 minHugeBlobInBytes)
457+
TDefragActor(const std::shared_ptr<TDefragCtx> &dCtx, const TIntrusivePtr<TBlobStorageGroupInfo> &info)
466458
: TActorBootstrapped<TDefragActor>()
467459
, DCtx(dCtx)
468460
, GInfo(info)
469-
, MinHugeBlobInBytes(minHugeBlobInBytes)
470461
{}
471462
};
472463

473464

474465
////////////////////////////////////////////////////////////////////////////
475466
// CreateDefragActor
476467
////////////////////////////////////////////////////////////////////////////
477-
IActor* CreateDefragActor(const std::shared_ptr<TDefragCtx> &dCtx, const TIntrusivePtr<TBlobStorageGroupInfo> &info,
478-
ui32 minHugeBlobInBytes) {
479-
return new TDefragActor(dCtx, info, minHugeBlobInBytes);
468+
IActor* CreateDefragActor(const std::shared_ptr<TDefragCtx> &dCtx, const TIntrusivePtr<TBlobStorageGroupInfo> &info) {
469+
return new TDefragActor(dCtx, info);
480470
}
481471

482472
} // NKikimr

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

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@ namespace NKikimr {
2525
const TActorId HugeKeeperId;
2626
NMonGroup::TDefragGroup DefragMonGroup;
2727
bool RunDefragBySchedule;
28-
const bool AddHeader;
2928

3029
// free up to this number of chunks in one quantum
3130
ui32 MaxChunksToDefrag = 1u;
@@ -61,7 +60,6 @@ namespace NKikimr {
6160
////////////////////////////////////////////////////////////////////////////
6261
IActor* CreateDefragActor(
6362
const std::shared_ptr<TDefragCtx> &dCtx,
64-
const TIntrusivePtr<TBlobStorageGroupInfo> &info,
65-
ui32 minHugeBlobInBytes);
63+
const TIntrusivePtr<TBlobStorageGroupInfo> &info);
6664

6765
} // NKikimr

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

Lines changed: 79 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ namespace NKikimr {
1919
std::shared_ptr<TDefragCtx> DCtx;
2020
const TVDiskID SelfVDiskId;
2121
std::optional<TChunksToDefrag> ChunksToDefrag;
22-
const ui32 MinHugeBlobInBytes;
2322

2423
enum {
2524
EvResume = EventSpaceBegin(TEvents::ES_PRIVATE)
@@ -29,12 +28,11 @@ namespace NKikimr {
2928

3029
public:
3130
TDefragQuantum(const std::shared_ptr<TDefragCtx>& dctx, const TVDiskID& selfVDiskId,
32-
std::optional<TChunksToDefrag> chunksToDefrag, ui32 minHugeBlobInBytes)
31+
std::optional<TChunksToDefrag> chunksToDefrag)
3332
: TActorCoroImpl(64_KB, true)
3433
, DCtx(dctx)
3534
, SelfVDiskId(selfVDiskId)
3635
, ChunksToDefrag(std::move(chunksToDefrag))
37-
, MinHugeBlobInBytes(minHugeBlobInBytes)
3836
{}
3937

4038
void ProcessUnexpectedEvent(TAutoPtr<IEventHandle> ev) {
@@ -66,8 +64,7 @@ namespace NKikimr {
6664
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD07, DCtx->VCtx->VDiskLogPrefix << "going to find chunks to defrag",
6765
(ActorId, SelfActorId));
6866

69-
TDefragQuantumFindChunks findChunks(GetSnapshot(), DCtx->HugeBlobCtx, ChunksToDefrag ?
70-
std::make_optional(std::move(ChunksToDefrag->ChunksToShred)) : std::nullopt);
67+
TDefragQuantumFindChunks findChunks(GetSnapshot(), DCtx->HugeBlobCtx);
7168
const ui64 endTime = GetCycleCountFast() + DurationToCycles(NDefrag::MaxSnapshotHoldDuration);
7269
while (findChunks.Scan(NDefrag::WorkQuantum)) {
7370
if (GetCycleCountFast() >= endTime) {
@@ -83,33 +80,67 @@ namespace NKikimr {
8380
Y_ABORT_UNLESS(*ChunksToDefrag || ChunksToDefrag->IsShred());
8481
}
8582
if (*ChunksToDefrag || ChunksToDefrag->IsShred()) {
86-
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD09, DCtx->VCtx->VDiskLogPrefix
87-
<< "commencing defragmentation", (ActorId, SelfActorId), (ChunksToDefrag, *ChunksToDefrag));
83+
const bool isShred = ChunksToDefrag->IsShred();
8884

89-
if (!ChunksToDefrag->IsShred()) {
90-
stat.FoundChunksToDefrag = ChunksToDefrag->FoundChunksToDefrag;
91-
stat.Eof = stat.FoundChunksToDefrag < DCtx->MaxChunksToDefrag;
85+
TDefragChunks lockedChunks;
9286

93-
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD10, DCtx->VCtx->VDiskLogPrefix << "locking chunks",
94-
(ActorId, SelfActorId));
87+
if (!isShred) {
88+
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD09, DCtx->VCtx->VDiskLogPrefix
89+
<< "commencing defragmentation", (ActorId, SelfActorId), (ChunksToDefrag, *ChunksToDefrag));
9590

96-
auto lockedChunks = LockChunks(*ChunksToDefrag);
97-
ChunksToDefrag->Chunks = std::move(lockedChunks);
91+
stat.FoundChunksToDefrag = ChunksToDefrag->FoundChunksToDefrag;
92+
stat.Eof = stat.FoundChunksToDefrag < DCtx->MaxChunksToDefrag;
9893
stat.FreedChunks = ChunksToDefrag->Chunks;
9994

95+
lockedChunks = LockChunks(*ChunksToDefrag);
96+
10097
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD11, DCtx->VCtx->VDiskLogPrefix << "locked chunks",
101-
(ActorId, SelfActorId), (LockedChunks, ChunksToDefrag->Chunks));
98+
(ActorId, SelfActorId), (LockedChunks, lockedChunks));
99+
} else {
100+
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD14, DCtx->VCtx->VDiskLogPrefix
101+
<< "commencing shredding", (ActorId, SelfActorId), (ChunksToShred, ChunksToDefrag->ChunksToShred));
102102
}
103103

104-
TDefragQuantumFindRecords findRecords(std::move(*ChunksToDefrag), DCtx->VCtx->Top->GType,
105-
DCtx->AddHeader, DCtx->HugeBlobCtx, MinHugeBlobInBytes);
104+
TDefragQuantumFindRecords findRecords(std::move(*ChunksToDefrag), lockedChunks);
106105
while (findRecords.Scan(NDefrag::WorkQuantum, GetSnapshot())) {
107106
Yield();
108107
}
109108

110109
if (auto records = findRecords.GetRecordsToRewrite(); !records.empty()) {
110+
THashMap<TChunkIdx, ui32> heatmap;
111+
for (const auto& record : records) {
112+
++heatmap[record.OldDiskPart.ChunkIdx];
113+
}
114+
115+
std::vector<std::tuple<TChunkIdx, ui32>> chunks(heatmap.begin(), heatmap.end());
116+
std::ranges::sort(chunks, std::less<ui32>(), [](const auto& x) { return std::get<1>(x); });
117+
118+
const size_t numRecordsTotal = records.size();
119+
120+
if (isShred && chunks.size() > DCtx->MaxChunksToDefrag) {
121+
chunks.resize(DCtx->MaxChunksToDefrag);
122+
THashSet<TChunkIdx> set;
123+
for (const auto& [chunkIdx, usage] : chunks) {
124+
set.insert(chunkIdx);
125+
}
126+
auto pred = [&](const auto& record) { return !set.contains(record.OldDiskPart.ChunkIdx); };
127+
auto range = std::ranges::remove_if(records, pred);
128+
records.erase(range.begin(), range.end());
129+
}
130+
131+
auto getSortedChunks = [&] {
132+
std::vector<TChunkIdx> temp;
133+
temp.reserve(chunks.size());
134+
for (const auto& [chunkIdx, usage] : chunks) {
135+
temp.push_back(chunkIdx);
136+
}
137+
std::ranges::sort(temp);
138+
return temp;
139+
};
140+
111141
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD12, DCtx->VCtx->VDiskLogPrefix << "rewriting records",
112-
(ActorId, SelfActorId), (NumRecordsToRewrite, records.size()));
142+
(ActorId, SelfActorId), (NumRecordsToRewrite, records.size()), (NumRecordsTotal, numRecordsTotal),
143+
(Chunks, getSortedChunks()));
113144

114145
const TActorId rewriterActorId = Register(CreateDefragRewriter(DCtx, SelfVDiskId, SelfActorId,
115146
std::move(records)));
@@ -122,19 +153,32 @@ namespace NKikimr {
122153
}
123154
stat.RewrittenRecs = ev->Get()->RewrittenRecs;
124155
stat.RewrittenBytes = ev->Get()->RewrittenBytes;
125-
} else if (ChunksToDefrag->IsShred()) {
126-
// no records to rewrite found
127-
stat.Eof = true;
156+
if (isShred) {
157+
stat.Eof = false;
158+
}
128159
}
129160

130-
auto tablesToCompact = std::move(findRecords.GetTablesToCompact());
161+
// scan index again to find tables we have to compact
162+
for (findRecords.StartFindingTablesToCompact(); findRecords.Scan(NDefrag::WorkQuantum, GetSnapshot()); Yield()) {}
163+
if (auto records = findRecords.GetRecordsToRewrite(); !records.empty()) {
164+
for (const auto& item : records) {
165+
STLOG(PRI_WARN, BS_VDISK_DEFRAG, BSVDD16, DCtx->VCtx->VDiskLogPrefix
166+
<< "blob found again after rewriting", (ActorId, SelfActorId), (Id, item.LogoBlobId),
167+
(Location, item.OldDiskPart));
168+
}
169+
}
131170

171+
auto tablesToCompact = findRecords.GetTablesToCompact();
172+
const bool needsFreshCompaction = findRecords.GetNeedsFreshCompaction();
132173
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD13, DCtx->VCtx->VDiskLogPrefix << "compacting",
133-
(ActorId, SelfActorId), (TablesToCompact, tablesToCompact));
134-
135-
Compact(tablesToCompact);
174+
(ActorId, SelfActorId), (TablesToCompact, tablesToCompact),
175+
(NeedsFreshCompaction, needsFreshCompaction));
176+
Compact(std::move(tablesToCompact), needsFreshCompaction);
136177
}
137178

179+
STLOG(PRI_DEBUG, BS_VDISK_DEFRAG, BSVDD15, DCtx->VCtx->VDiskLogPrefix << "quantum finished",
180+
(ActorId, SelfActorId), (Stat, stat));
181+
138182
Send(ParentActorId, new TEvDefragQuantumResult(std::move(stat)));
139183
}
140184

@@ -154,15 +198,21 @@ namespace NKikimr {
154198
return res->Get()->LockedChunks;
155199
}
156200

157-
void Compact(THashSet<ui64> tablesToCompact) {
158-
Send(DCtx->SkeletonId, TEvCompactVDisk::Create(EHullDbType::LogoBlobs, tablesToCompact));
201+
void Compact(THashSet<ui64> tablesToCompact, bool needsFreshCompaction) {
202+
if (tablesToCompact) {
203+
Send(DCtx->SkeletonId, TEvCompactVDisk::Create(EHullDbType::LogoBlobs, std::move(tablesToCompact)));
204+
} else if (needsFreshCompaction) {
205+
Send(DCtx->SkeletonId, TEvCompactVDisk::Create(EHullDbType::LogoBlobs, TEvCompactVDisk::EMode::FRESH_ONLY));
206+
} else {
207+
return; // nothing to do
208+
}
159209
WaitForSpecificEvent<TEvCompactVDiskResult>(&TDefragQuantum::ProcessUnexpectedEvent);
160210
}
161211
};
162212

163213
IActor *CreateDefragQuantumActor(const std::shared_ptr<TDefragCtx>& dctx, const TVDiskID& selfVDiskId,
164-
std::optional<TChunksToDefrag> chunksToDefrag, ui32 minHugeBlobInBytes) {
165-
return new TActorCoro(MakeHolder<TDefragQuantum>(dctx, selfVDiskId, std::move(chunksToDefrag), minHugeBlobInBytes),
214+
std::optional<TChunksToDefrag> chunksToDefrag) {
215+
return new TActorCoro(MakeHolder<TDefragQuantum>(dctx, selfVDiskId, std::move(chunksToDefrag)),
166216
NKikimrServices::TActivity::BS_DEFRAG_QUANTUM);
167217
}
168218

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ namespace NKikimr {
4444
struct TChunksToDefrag;
4545

4646
IActor *CreateDefragQuantumActor(const std::shared_ptr<TDefragCtx>& dctx, const TVDiskID& selfVDiskId,
47-
std::optional<TChunksToDefrag> chunksToDefrag, ui32 minHugeBlobInBytes);
47+
std::optional<TChunksToDefrag> chunksToDefrag);
4848

4949
} // NKikimr
5050

0 commit comments

Comments
 (0)