Skip to content

Commit d941351

Browse files
kungakunga
authored
Forward cache tests (#2346)
1 parent 11dd3f3 commit d941351

File tree

8 files changed

+456
-86
lines changed

8 files changed

+456
-86
lines changed

‎ydb/core/tablet_flat/flat_fwd_cache.h

Lines changed: 51 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -10,52 +10,51 @@ namespace NKikimr {
1010
namespace NTable {
1111
namespace NFwd {
1212

13-
class TCache : public IPageLoadingLogic {
14-
using TGroupId = NPage::TGroupId;
15-
16-
template<size_t Items>
17-
struct TRound {
18-
const TSharedData* Get(TPageId pageId) const
19-
{
20-
if (pageId < Edge) {
21-
const auto pred = [pageId](const NPageCollection::TLoadedPage &page) {
22-
return page.PageId == pageId;
23-
};
24-
25-
auto it = std::find_if(Pages.begin(), Pages.end(), pred);
26-
27-
if (it == Pages.end()) {
28-
Y_ABORT("Failed to locate page within forward trace");
13+
template<size_t Capacity>
14+
class TLoadedPagesCircularBuffer {
15+
public:
16+
const TSharedData* Get(TPageId pageId) const
17+
{
18+
if (pageId < FirstUnseenPageId) {
19+
for (const auto& page : LoadedPages) {
20+
if (page.PageId == pageId) {
21+
return &page.Data;
2922
}
30-
31-
return &it->Data;
3223
}
3324

34-
return nullptr;
25+
Y_ABORT("Failed to locate page within forward trace");
3526
}
3627

37-
ui32 Emplace(TPage &page)
38-
{
39-
Y_ABORT_UNLESS(page, "Cannot push invalid page to trace cache");
28+
// next pages may be requested, ignore them
29+
return nullptr;
30+
}
31+
32+
// returns released data size
33+
ui64 Emplace(TPage &page)
34+
{
35+
Y_ABORT_UNLESS(page, "Cannot push invalid page to trace cache");
4036

41-
Offset = (Pages.size() + Offset - 1) % Pages.size();
37+
Offset = (Offset + 1) % Capacity;
4238

43-
const ui32 was = Pages[Offset].Data.size();
39+
const ui64 releasedDataSize = LoadedPages[Offset].Data.size();
4440

45-
Pages[Offset].Data = page.Release();
46-
Pages[Offset].PageId = page.PageId;
47-
Edge = Max(Edge, page.PageId + 1);
41+
LoadedPages[Offset].Data = page.Release();
42+
LoadedPages[Offset].PageId = page.PageId;
43+
FirstUnseenPageId = Max(FirstUnseenPageId, page.PageId + 1);
4844

49-
return was;
50-
}
45+
return releasedDataSize;
46+
}
5147

52-
private:
53-
std::array<NPageCollection::TLoadedPage, Items> Pages;
54-
TPageId Edge = 0;
55-
ui32 Offset = 0;
56-
};
48+
private:
49+
std::array<NPageCollection::TLoadedPage, Capacity> LoadedPages;
50+
ui32 Offset = 0;
51+
TPageId FirstUnseenPageId = 0;
52+
};
5753

54+
class TCache : public IPageLoadingLogic {
5855
public:
56+
using TGroupId = NPage::TGroupId;
57+
5958
TCache() = delete;
6059

6160
TCache(const TPart* part, IPages* env, TGroupId groupId, const TIntrusiveConstPtr<TSlices>& bounds = nullptr)
@@ -64,17 +63,21 @@ namespace NFwd {
6463

6564
~TCache()
6665
{
67-
for (auto &it: Pages) it.Release();
66+
for (auto &it: Pages) {
67+
it.Release();
68+
}
6869
}
6970

7071
TResult Handle(IPageLoadingQueue *head, TPageId pageId, ui64 lower) noexcept override
7172
{
7273
Y_ABORT_UNLESS(pageId != Max<TPageId>(), "Invalid requested pageId");
7374

74-
if (auto *page = Trace.Get(pageId))
75+
if (auto *page = Trace.Get(pageId)) {
7576
return { page, false, true };
77+
}
7678

77-
Rewind(pageId).Shrink(); /* points Offset to pageId */
79+
Rewind(pageId); /* points Offset to pageId */
80+
Shrink();
7881

7982
bool more = Grow && (OnHold + OnFetch <= lower);
8083

@@ -105,7 +108,7 @@ namespace NFwd {
105108

106109
Stat.Saved += one.Data.size();
107110
OnFetch -= one.Data.size();
108-
OnHold += it->Settle(one);
111+
OnHold += it->Settle(one); // settle of a dropped page returns 0 and releases its data
109112

110113
++it;
111114
}
@@ -135,7 +138,7 @@ namespace NFwd {
135138
return Pages.at(Offset);
136139
}
137140

138-
TCache& Rewind(TPageId pageId) noexcept
141+
void Rewind(TPageId pageId) noexcept
139142
{
140143
while (auto drop = Index.Clean(pageId)) {
141144
auto &page = Pages.at(Offset);
@@ -144,32 +147,29 @@ namespace NFwd {
144147
Y_ABORT("Dropping page that is not exist in cache");
145148
} else if (page.Size == 0) {
146149
Y_ABORT("Dropping page that has not been touched");
147-
} else if (page.Usage == EUsage::Keep) {
150+
} else if (page.Usage == EUsage::Keep && page) {
148151
OnHold -= Trace.Emplace(page);
149-
} else if (auto size = page.Release().size()) {
150-
OnHold -= size;
151-
152-
*(page.Ready() ? &Stat.After : &Stat.Before) += size;
152+
} else {
153+
OnHold -= page.Release().size();
154+
*(page.Ready() ? &Stat.After : &Stat.Before) += page.Size;
153155
}
154156

157+
// keep pending pages but increment offset
155158
Offset++;
156159
}
157-
158-
return *this;
159160
}
160161

161-
TCache& Shrink() noexcept
162+
void Shrink() noexcept
162163
{
163-
for (; Offset && Pages[0].Ready(); Offset--)
164+
for (; Offset && Pages[0].Ready(); Offset--) {
164165
Pages.pop_front();
165-
166-
return *this;
166+
}
167167
}
168168

169169
private:
170170
bool Grow = true; /* Have some pages for Forward(...) */
171171
TForward Index;
172-
TRound<TPart::Trace> Trace;
172+
TLoadedPagesCircularBuffer<TPart::Trace> Trace;
173173

174174
/*_ Forward cache line state */
175175

‎ydb/core/tablet_flat/flat_fwd_misc.cpp

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
#include "flat_fwd_misc.h"
2+
3+
Y_DECLARE_OUT_SPEC(, NKikimr::NTable::NFwd::TStat, stream, value) {
4+
value.Describe(stream);
5+
}

‎ydb/core/tablet_flat/flat_fwd_misc.h

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,11 @@ namespace NFwd {
1818
{
1919
out
2020
<< "TFwd{"
21-
<< Fetch << "b"
22-
<< " > " << Saved << "b"
23-
<< " > " << Usage << "b"
24-
<< " +" << After << "b"
25-
<< " ~" << Before << "b"
21+
<< "fetch=" << Fetch
22+
<< ",saved=" << Saved
23+
<< ",usage=" << Usage
24+
<< ",after=" << After
25+
<< ",before=" << Before
2626
<< "}";
2727
}
2828

@@ -37,6 +37,8 @@ namespace NFwd {
3737
return *this;
3838
}
3939

40+
auto operator<=>(const TStat&) const = default;
41+
4042
ui64 Fetch = 0; /* Requested to load by cache */
4143
ui64 Saved = 0; /* Obtained by cache with DoSave() */
4244
ui64 Usage = 0; /* Actually was used by client */

‎ydb/core/tablet_flat/flat_fwd_page.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ namespace NFwd {
7575
const TSharedData* Touch(TPageId pageId, TStat &stat) noexcept
7676
{
7777
if (PageId != pageId || (!Data && Fetch == EFetch::Done)) {
78-
Y_ABORT("Touching page thatd doesn't fits to this action");
78+
Y_ABORT("Touching page that doesn't fit to this action");
7979
} else {
8080
auto to = Fetch == EFetch::None ? EUsage::Seen : EUsage::Keep;
8181

‎ydb/core/tablet_flat/flat_page_other.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ namespace NPage {
2828
Tags.resize(TagsCount + (TagsCount & 1), 0);
2929
Cook.reserve(tags);
3030

31-
Y_ABORT_UNLESS(tags <= ui32(-Min<i16>()), "Too many columnt tags");
31+
Y_ABORT_UNLESS(tags <= ui32(-Min<i16>()), "Too many column tags");
3232
}
3333

3434
void Put(TRowId row, ui16 tag, ui32 bytes) noexcept

‎ydb/core/tablet_flat/test/libs/table/test_part.h

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -177,6 +177,18 @@ namespace NTest {
177177
return index.GetLastRecord();
178178
}
179179

180+
inline TRowId GetPageId(const TPart& part, ui32 pageIndex) {
181+
TTestEnv env;
182+
TPartIndexIt index(&part, &env, { });
183+
184+
Y_ABORT_UNLESS(index.Seek(0) == EReady::Data);
185+
for (TPageId p = 0; p < pageIndex; p++) {
186+
Y_ABORT_UNLESS(index.Next() == EReady::Data);
187+
}
188+
189+
return index.GetPageId();
190+
}
191+
180192
inline TRowId GetRowId(const TPart& part, ui32 pageIndex) {
181193
TTestEnv env;
182194
TPartIndexIt index(&part, &env, { });

0 commit comments

Comments
 (0)