Skip to content

Remove share body logic #11987

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Nov 28, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions ydb/core/tablet_flat/flat_executor.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -2771,9 +2771,6 @@ void TExecutor::Handle(NSharedCache::TEvUpdated::TPtr &ev) {

for (auto &kv : msg->Actions) {
if (auto *info = PrivatePageCache->Info(kv.first)) {
for (auto &kvCorrected : kv.second.Accepted) {
PrivatePageCache->UpdateSharedBody(info, kvCorrected.first, std::move(kvCorrected.second));
}
for (ui32 pageId : kv.second.Dropped) {
PrivatePageCache->DropSharedBody(info, pageId);
}
Expand Down Expand Up @@ -3598,7 +3595,6 @@ void TExecutor::UpdateCounters(const TActorContext &ctx) {
Counters->Simple()[TExecutorCounters::CACHE_TOTAL_SHARED_BODY].Set(stats.TotalSharedBody);
Counters->Simple()[TExecutorCounters::CACHE_TOTAL_PINNED_BODY].Set(stats.TotalPinnedBody);
Counters->Simple()[TExecutorCounters::CACHE_TOTAL_EXCLUSIVE].Set(stats.TotalExclusive);
Counters->Simple()[TExecutorCounters::CACHE_TOTAL_SHARED_PENDING].Set(stats.TotalSharedPending);
Counters->Simple()[TExecutorCounters::CACHE_TOTAL_STICKY].Set(stats.TotalSticky);
}

Expand Down Expand Up @@ -4119,7 +4115,6 @@ void TExecutor::RenderHtmlPage(NMon::TEvRemoteHttpInfo::TPtr &ev) const {
DIV_CLASS("row") {str << "Total bytes in shared cache: " << PrivatePageCache->GetStats().TotalSharedBody; }
DIV_CLASS("row") {str << "Total bytes in local cache: " << PrivatePageCache->GetStats().TotalPinnedBody; }
DIV_CLASS("row") {str << "Total bytes exclusive to local cache: " << PrivatePageCache->GetStats().TotalExclusive; }
DIV_CLASS("row") {str << "Total bytes in transit to shared cache: " << PrivatePageCache->GetStats().TotalSharedPending; }
DIV_CLASS("row") {str << "Total bytes marked as sticky: " << PrivatePageCache->GetStats().TotalSticky; }

if (GcLogic) {
Expand Down
4 changes: 2 additions & 2 deletions ydb/core/tablet_flat/flat_executor_tx_env.h
Original file line number Diff line number Diff line change
Expand Up @@ -38,11 +38,11 @@ namespace NTabletFlatExecutor {
return { !ReadMissingReferences, page };
}

const TSharedData* TryGetPage(const TPart* part, TPageId page, TGroupId groupId) override
const TSharedData* TryGetPage(const TPart* part, TPageId pageId, TGroupId groupId) override
{
auto *partStore = CheckedCast<const NTable::TPartStore*>(part);

return Lookup(partStore->PageCollections.at(groupId.Index).Get(), page);
return Lookup(partStore->PageCollections.at(groupId.Index).Get(), pageId);
}

void EnableReadMissingReferences() noexcept {
Expand Down
73 changes: 7 additions & 66 deletions ydb/core/tablet_flat/flat_sausagecache.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@ namespace NTabletFlatExecutor {
TPrivatePageCache::TPage::TPage(size_t size, TPageId pageId, TInfo* info)
: LoadState(LoadStateNo)
, Sticky(false)
, SharedPending(false)
, Id(pageId)
, Size(size)
, Info(info)
Expand Down Expand Up @@ -47,20 +46,15 @@ void TPrivatePageCache::RegisterPageCollection(TIntrusivePtr<TInfo> info) {

for (const auto& kv : info->PageMap) {
auto* page = kv.second.Get();
Y_DEBUG_ABORT_UNLESS(page);

if (page->SharedBody)
Stats.TotalSharedBody += page->Size;
Y_ABORT_UNLESS(page);
Y_ABORT_UNLESS(page->SharedBody, "New filled pages can't be without a shared body");

Stats.TotalSharedBody += page->Size;
if (page->PinnedBody)
Stats.TotalPinnedBody += page->Size;
if (page->PinnedBody && !page->SharedBody)
Stats.TotalExclusive += page->Size;
if (page->Sticky)
Stats.TotalSticky += page->Size;

Y_DEBUG_ABORT_UNLESS(!page->SharedPending, "New page shouldn't be shared pending");
TryShareBody(page);

TryUnload(page);
Y_DEBUG_ABORT_UNLESS(!page->IsUnnecessary());
}
Expand Down Expand Up @@ -132,8 +126,6 @@ bool TPrivatePageCache::UnlockPageCollection(TLogoBlobID id) {
Stats.TotalPinnedBody -= page->Size;
if (page->PinnedBody && !page->SharedBody)
Stats.TotalExclusive -= page->Size;
if (page->SharedPending)
Stats.TotalSharedPending -= page->Size;
if (page->Sticky)
Stats.TotalSticky -= page->Size;
}
Expand Down Expand Up @@ -204,8 +196,6 @@ std::pair<ui32, ui64> TPrivatePageCache::Request(TVector<ui32> &pages, TPrivateP
TPage *page = info->EnsurePage(*it);
switch (page->LoadState) {
case TPage::LoadStateNo:
Y_DEBUG_ABORT_UNLESS(!page->SharedPending, "Trying to load a page that may be restored");
[[fallthrough]];
case TPage::LoadStateRequestedAsync:
page->LoadState = TPage::LoadStateRequested;
bytesToRequest += page->Size;
Expand Down Expand Up @@ -261,8 +251,8 @@ void TPrivatePageCache::TryLoad(TPage *page) {

void TPrivatePageCache::TPrivatePageCache::TryUnload(TPage *page) {
if (page->LoadState == TPage::LoadStateLoaded) {
if (!page->SharedPending && !page->PinPad && !page->Sticky) {
ToTouchShared[page->Info->Id][page->Id];
if (!page->PinPad && !page->Sticky) {
ToTouchShared[page->Info->Id].insert(page->Id);
page->LoadState = TPage::LoadStateNo;
if (Y_LIKELY(page->PinnedBody)) {
Stats.TotalPinnedBody -= page->Size;
Expand Down Expand Up @@ -291,19 +281,6 @@ void TPrivatePageCache::TPrivatePageCache::TryEraseIfUnnecessary(TPage *page) {
}
}

void TPrivatePageCache::TPrivatePageCache::TryShareBody(TPage *page) {
if (page->LoadState == TPage::LoadStateLoaded) {
auto &x = ToTouchShared[page->Info->Id][page->Id];
if (!page->SharedPending && !page->SharedBody && page->PinnedBody) {
// We keep pinned body around until it's either
// accepted or dropped by the shared cache
page->SharedPending = true;
Stats.TotalSharedPending += page->Size;
x = page->PinnedBody;
}
}
}

const TSharedData* TPrivatePageCache::Lookup(TPageId pageId, TInfo *info) {
TPage *page = info->EnsurePage(pageId);

Expand Down Expand Up @@ -502,45 +479,11 @@ void TPrivatePageCache::ResetTouchesAndToLoad(bool verifyEmpty) {
Stats.CurrentCacheMisses = 0;
}

void TPrivatePageCache::UpdateSharedBody(TInfo *info, TPageId pageId, TSharedPageRef shared) {
TPage *page = info->GetPage(pageId);
if (!page)
return;

// Note: shared cache may accept a pending page if it is used by multiple private caches
// (for example, used by tablet and its follower)
if (Y_UNLIKELY(!page->SharedPending)) {
return;
}
Y_DEBUG_ABORT_UNLESS(page->LoadState == TPage::LoadStateLoaded, "Shared pending page should be loaded");

// Shared cache accepted our page and provided its shared reference
Stats.TotalSharedPending -= page->Size;
page->SharedPending = false;
if (Y_LIKELY(!page->SharedBody)) {
Stats.TotalSharedBody += page->Size;
if (Y_LIKELY(page->PinnedBody)) {
Stats.TotalExclusive -= page->Size;
}
}
page->SharedBody = std::move(shared);
TryUnload(page);
}

void TPrivatePageCache::DropSharedBody(TInfo *info, TPageId pageId) {
TPage *page = info->GetPage(pageId);
if (!page)
return;

// Note: shared cache may drop a pending page if it is used by multiple private caches
// (for example, used by tablet and its follower)
if (Y_UNLIKELY(page->SharedPending)) {
// Shared cache rejected our page so we should drop it too
Stats.TotalSharedPending -= page->Size;
page->SharedPending = false;
TryUnload(page);
}

if (!page->SharedBody.IsUsed()) {
if (Y_LIKELY(page->SharedBody)) {
Stats.TotalSharedBody -= page->Size;
Expand Down Expand Up @@ -568,8 +511,6 @@ TPrivatePageCache::TPage::TWaitQueuePtr TPrivatePageCache::ProvideBlock(
Stats.TotalPinnedBody -= page->Size;
if (Y_UNLIKELY(page->PinnedBody && !page->SharedBody))
Stats.TotalExclusive -= page->Size;
if (Y_UNLIKELY(page->SharedPending))
Stats.TotalSharedPending -= page->Size;

// Note: we must be careful not to accidentally drop the sticky bit
page->Fill(std::move(loaded.Page), page->Sticky);
Expand Down Expand Up @@ -603,7 +544,7 @@ THashMap<TLogoBlobID, TIntrusivePtr<TPrivatePageCache::TInfo>> TPrivatePageCache
return ret;
}

THashMap<TLogoBlobID, THashMap<TPrivatePageCache::TPageId, TSharedData>> TPrivatePageCache::GetPrepareSharedTouched() {
THashMap<TLogoBlobID, THashSet<TPrivatePageCache::TPageId>> TPrivatePageCache::GetPrepareSharedTouched() {
return std::move(ToTouchShared);
}

Expand Down
19 changes: 3 additions & 16 deletions ydb/core/tablet_flat/flat_sausagecache.h
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@ class TPrivatePageCache {
ui64 TotalSharedBody = 0; // total number of bytes currently referenced from shared cache
ui64 TotalPinnedBody = 0; // total number of bytes currently pinned in memory
ui64 TotalExclusive = 0; // total number of bytes exclusive to this cache (not from shared cache)
ui64 TotalSharedPending = 0; // total number of bytes waiting for transfer to shared cache
ui64 TotalSticky = 0; // total number of bytes marked as sticky (never unloaded from memory)
ui64 PinnedSetSize = 0; // number of bytes pinned by transactions (even those not currently loaded)
ui64 PinnedLoadSize = 0; // number of bytes pinned by transactions (which are currently being loaded)
Expand All @@ -54,7 +53,6 @@ class TPrivatePageCache {

ui32 LoadState : 2;
ui32 Sticky : 1;
ui32 SharedPending : 1;

const TPageId Id;
const size_t Size;
Expand All @@ -74,22 +72,13 @@ class TPrivatePageCache {
return (
LoadState == LoadStateNo &&
!Sticky &&
!SharedPending &&
!PinPad &&
!WaitQueue &&
!SharedBody);
}

void Fill(TSharedData data, bool sticky = false) {
Y_DEBUG_ABORT_UNLESS(!SharedBody && !SharedPending, "Populating cache with shared data already present");
void Fill(TSharedPageRef shared, bool sticky) {
Sticky = sticky;
LoadState = LoadStateLoaded;
PinnedBody = std::move(data);
}

void Fill(TSharedPageRef shared, bool sticky = false) {
Sticky = sticky;
SharedPending = false;
SharedBody = std::move(shared);
LoadState = LoadStateLoaded;
PinnedBody = TPinnedPageRef(SharedBody).GetData();
Expand Down Expand Up @@ -170,16 +159,15 @@ class TPrivatePageCache {
THashMap<TPrivatePageCache::TInfo*, TVector<ui32>> GetToLoad() const;
void ResetTouchesAndToLoad(bool verifyEmpty);

void UpdateSharedBody(TInfo *collectionInfo, TPageId pageId, TSharedPageRef shared);
void DropSharedBody(TInfo *collectionInfo, TPageId pageId);

TPage::TWaitQueuePtr ProvideBlock(NSharedCache::TEvResult::TLoaded&& loaded, TInfo *collectionInfo);
THashMap<TLogoBlobID, TIntrusivePtr<TInfo>> DetachPrivatePageCache();
THashMap<TLogoBlobID, THashMap<TPageId, TSharedData>> GetPrepareSharedTouched();
THashMap<TLogoBlobID, THashSet<TPageId>> GetPrepareSharedTouched();

private:
THashMap<TLogoBlobID, TIntrusivePtr<TInfo>> PageCollections;
THashMap<TLogoBlobID, THashMap<TPageId, TSharedData>> ToTouchShared;
THashMap<TLogoBlobID, THashSet<TPageId>> ToTouchShared;

TStats Stats;

Expand All @@ -192,7 +180,6 @@ class TPrivatePageCache {
void TryLoad(TPage *page);
void TryUnload(TPage *page);
void TryEraseIfUnnecessary(TPage *page);
void TryShareBody(TPage *page);
};

}}
6 changes: 3 additions & 3 deletions ydb/core/tablet_flat/shared_cache_events.h
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

namespace NKikimr::NSharedCache {
using EPriority = NTabletFlatExecutor::NBlockIO::EPriority;
using TPageId = NTable::NPage::TPageId;

enum EEv {
EvBegin = EventSpaceBegin(TKikimrEvents::ES_FLAT_EXECUTOR),
Expand Down Expand Up @@ -45,9 +46,9 @@ namespace NKikimr::NSharedCache {
};

struct TEvTouch : public TEventLocal<TEvTouch, EvTouch> {
THashMap<TLogoBlobID, THashMap<ui32, TSharedData>> Touched;
THashMap<TLogoBlobID, THashSet<TPageId>> Touched;

TEvTouch(THashMap<TLogoBlobID, THashMap<ui32, TSharedData>> &&touched)
TEvTouch(THashMap<TLogoBlobID, THashSet<TPageId>> &&touched)
: Touched(std::move(touched))
{}
};
Expand Down Expand Up @@ -135,7 +136,6 @@ namespace NKikimr::NSharedCache {

struct TEvUpdated : public TEventLocal<TEvUpdated, EvUpdated> {
struct TActions {
THashMap<ui32, TSharedPageRef> Accepted;
THashSet<ui32> Dropped;
};

Expand Down
46 changes: 8 additions & 38 deletions ydb/core/tablet_flat/shared_sausagecache.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -709,41 +709,17 @@ class TSharedPageCache : public TActorBootstrapped<TSharedPageCache> {

void Handle(NSharedCache::TEvTouch::TPtr &ev) {
NSharedCache::TEvTouch *msg = ev->Get();
THashMap<TLogoBlobID, NSharedCache::TEvUpdated::TActions> actions;
for (auto &xpair : msg->Touched) {
auto collectionIt = Collections.find(xpair.first);
if (collectionIt == Collections.end()) {
for (auto &x : xpair.second) {
if (x.second) {
actions[xpair.first].Dropped.insert(x.first);
x.second = { };
}
}

for (auto &[metaId, touchedPages] : msg->Touched) {
auto collection = Collections.FindPtr(metaId);
if (!collection) {
continue;
}
auto &collection = collectionIt->second;
for (auto &x : xpair.second) {
const ui32 pageId = x.first;
Y_ABORT_UNLESS(pageId < collection.PageMap.size());
auto* page = collection.PageMap[pageId].Get();
for (auto pageId : touchedPages) {
Y_ABORT_UNLESS(pageId < collection->PageMap.size());
auto* page = collection->PageMap[pageId].Get();
if (!page) {
if (x.second) {
Y_ABORT_UNLESS(collection.PageMap.emplace(pageId, (page = new TPage(pageId, x.second.size(), &collection))));
} else {
continue;
}
}
Y_ABORT_UNLESS(page);

if (auto body = std::move(x.second)) {
if (page->HasMissingBody()) {
page->Initialize(std::move(body));
BodyProvided(collection, page);
}

auto ref = TSharedPageRef::MakeUsed(page, SharedCachePages->GCList);
Y_ABORT_UNLESS(ref.IsUsed(), "Unexpected failure to grab a cached page");
actions[xpair.first].Accepted[pageId] = std::move(ref);
continue;
}

switch (page->State) {
Expand All @@ -768,12 +744,6 @@ class TSharedPageCache : public TActorBootstrapped<TSharedPageCache> {
}
}

if (actions) {
auto msg = MakeHolder<NSharedCache::TEvUpdated>();
msg->Actions = std::move(actions);
Send(ev->Sender, msg.Release());
}

DoGC();
}

Expand Down
Loading
Loading