Remove NPageCollection::TFetch class (#21476)

Author: kunga

ydb/core/tablet_flat/flat_bio_actor.cpp

@@ -14,10 +14,10 @@ using TEvGet = TEvBlobStorage::TEvGet;
 
 struct TBlockIO::TLoaded : public TEvBlobStorage::TEvGetResult::TResponse{ };
 
-TBlockIO::TBlockIO(TActorId service, ui64 cookie)
+TBlockIO::TBlockIO(TActorId statActorId, ui64 eventCookie)
     : ::NActors::IActorCallback(static_cast<TReceiveFunc>(&TBlockIO::Inbox), NKikimrServices::TActivity::SAUSAGE_BIO_A)
-    , Service(service)
-    , Cookie(cookie)
+    , StatActorId(statActorId)
+    , EventCookie(eventCookie)
 {
 }
 
@@ -42,10 +42,24 @@ void TBlockIO::Inbox(TEventHandlePtr &eh)
             Handle(eh->Cookie, { ptr, size_t(ev->ResponseSz) });
         }
     } else if (auto *ev = eh->CastAsLocal<NBlockIO::TEvFetch>()) {
-        Y_ENSURE(!Owner, "TBlockIO actor now can hanle only one request");
+        Y_ENSURE(!Sender, "TBlockIO actor now can handle only one request");
+        Sender = eh->Sender;
+
+        Priority = ev->Priority;
+        TraceId = std::move(ev->TraceId);
+        RequestCookie = ev->Cookie;
+
+        PageCollection = std::move(ev->PageCollection);
+        Pages = std::move(ev->Pages);
+        Y_ENSURE(Pages, "Got TFetch request without pages list");
+        PagesToBlobsConverter = new TPagesToBlobsConverter(*PageCollection, Pages);
+        BlockStates.reserve(Pages.size());
+        for (auto page: Pages) {
+            ui64 size = PageCollection->Page(page).Size;
+            BlockStates.emplace_back(size);
+        }
 
-        Owner = eh->Sender;
-        Bootstrap(ev->Priority, ev->Fetch);
+        Dispatch();
     } else if (eh->CastAsLocal<TEvents::TEvUndelivered>()) {
         Terminate(NKikimrProto::UNKNOWN);
     } else if (eh->CastAsLocal<TEvents::TEvPoison>()) {
@@ -55,25 +69,6 @@ void TBlockIO::Inbox(TEventHandlePtr &eh)
     }
 }
 
-void TBlockIO::Bootstrap(EPriority priority, TAutoPtr<NPageCollection::TFetch> origin)
-{
-    Origin = origin;
-    Priority = priority;
-
-    Y_ENSURE(Origin->Pages, "Got TFetch request without pages list");
-
-    PagesToBlobsConverter = new TPagesToBlobsConverter(*Origin->PageCollection, Origin->Pages);
-
-    BlockStates.reserve(Origin->Pages.size());
-
-    for (auto page: Origin->Pages) {
-        ui64 size = Origin->PageCollection->Page(page).Size;
-        BlockStates.emplace_back(size);
-    }
-
-    Dispatch();
-}
-
 void TBlockIO::Dispatch()
 {
     const auto ctx = ActorContext();
@@ -101,7 +96,7 @@ void TBlockIO::Dispatch()
         ui32 lastBlob = Max<ui32>();
         for (const auto on : xrange(+more)) {
             auto &brick = PagesToBlobsConverter->Queue[more.From + on];
-            auto glob = Origin->PageCollection->Glob(brick.Blob);
+            auto glob = PageCollection->Glob(brick.Blob);
 
             if ((group = (on ? group : glob.Group)) != glob.Group) {
                 Y_TABLET_ERROR("Cannot handle different groups in one request");
@@ -127,12 +122,12 @@ void TBlockIO::Dispatch()
 
         auto *ev = new TEvGet(query, +more, TInstant::Max(), klass, false);
 
-        SendToBSProxy(ctx, group, ev, more.From /* cookie, request offset */, std::move(Origin->TraceId));
+        SendToBSProxy(ctx, group, ev, more.From /* cookie, request offset */, std::move(TraceId));
     }
 
     if (auto logl = Logger->Log(ELnLev::Debug)) {
         logl
-            << "NBlockIO pageCollection " << Origin->PageCollection->Label() << " cooked flow "
+            << "NBlockIO pageCollection " << PageCollection->Label() << " cooked flow "
             << PagesToBlobsConverter->OnHold << "b " << PagesToBlobsConverter->Tail << "p" << " " << PagesToBlobsConverter->Queue.size()
             << " bricks in " << Pending << " reads, " << BlockStates.size() <<  "p req";
     }
@@ -144,15 +139,15 @@ void TBlockIO::Handle(ui32 base, TArrayRef<TLoaded> items)
 {
     if (auto logl = Logger->Log(ELnLev::Debug)) {
         logl
-            << "NBlockIO pageCollection " << Origin->PageCollection->Label() << " got base"
+            << "NBlockIO pageCollection " << PageCollection->Label() << " got base"
             << " " << items.size() << " bricks, left " << Pending;
     }
 
     for (auto &piece: items) {
         if (piece.Status != NKikimrProto::OK) {
             if (auto logl = Logger->Log(ELnLev::Warn)) {
                 logl
-                    << "NBlockIO pageCollection " << Origin->PageCollection->Label() << " get failed"
+                    << "NBlockIO pageCollection " << PageCollection->Label() << " get failed"
                     << ", " << piece.Id << " status " << piece.Status;
             }
 
@@ -171,21 +166,21 @@ void TBlockIO::Handle(ui32 base, TArrayRef<TLoaded> items)
         return;
 
     size_t index = 0;
-    for (ui32 pageId : Origin->Pages) {
+    for (auto pageId : Pages) {
         auto& state = BlockStates.at(index++);
         Y_ENSURE(state.Offset == state.Data.size());
-        if (Origin->PageCollection->Verify(pageId, state.Data)) {
+        if (PageCollection->Verify(pageId, state.Data)) {
             continue;
         } else if (auto logl = Logger->Log(ELnLev::Crit)) {
-            const auto bnd = Origin->PageCollection->Bounds(pageId);
+            const auto bnd = PageCollection->Bounds(pageId);
 
             logl
-                << "NBlockIO pageCollection " << Origin->PageCollection->Label() << " verify failed"
+                << "NBlockIO pageCollection " << PageCollection->Label() << " verify failed"
                 << ", page " << pageId << " " << state.Data.size() << "b"
                 << " spans over {";
 
             for (auto one: xrange(bnd.Lo.Blob, bnd.Up.Blob + 1)) {
-                const auto glob = Origin->PageCollection->Glob(one);
+                const auto glob = PageCollection->Glob(one);
 
                 logl << " " << glob.Group << " " << glob.Logo;
             }
@@ -203,26 +198,26 @@ void TBlockIO::Terminate(EStatus code)
 {
     if (auto logl = Logger->Log(code ? ELnLev::Warn : ELnLev::Debug)) {
         logl
-            << "NBlockIO pageCollection " << Origin->PageCollection->Label() << " end, status " << code
-            << ", cookie {req " << Origin->Cookie << " ev " << Cookie << "}"
+            << "NBlockIO pageCollection " << PageCollection->Label() << " end, status " << code
+            << ", cookie {req " << RequestCookie << " ev " << EventCookie << "}"
             << ", " << BlockStates.size() << " pages";
     }
 
-    auto *ev = new TEvData(Origin, code);
+    auto *ev = new TEvData(code, PageCollection, RequestCookie);
 
-    if (code == NKikimrProto::OK) {
-        size_t index = 0;
-        ev->Blocks.reserve(ev->Fetch->Pages.size());
-        for (ui32 pageId : ev->Fetch->Pages) {
-            auto& state = BlockStates.at(index++);
-            ev->Blocks.emplace_back(pageId, std::move(state.Data));
+    ev->Pages.resize(Pages.size());
+    for (auto index : xrange(Pages.size())) {
+        auto& page = ev->Pages[index];
+        page.PageId = Pages[index];
+        if (code == NKikimrProto::OK) {
+            page.Data = std::move(BlockStates.at(index++).Data);
         }
     }
 
-    if (Service)
-        Send(Service, new TEvStat(EDir::Read, Priority, PagesToBlobsConverter->OnHold, TotalOps, std::move(GroupBytes), std::move(GroupOps)));
+    if (StatActorId)
+        Send(StatActorId, new TEvStat(EDir::Read, Priority, PagesToBlobsConverter->OnHold, TotalOps, std::move(GroupBytes), std::move(GroupOps)));
 
-    Send(Owner, ev, 0, Cookie);
+    Send(Sender, ev, 0, EventCookie);
 
     return PassAway();
 }

ydb/core/tablet_flat/flat_bio_actor.h

@@ -24,21 +24,23 @@ namespace NBlockIO {
     private:
         void Registered(TActorSystem*, const TActorId&) override;
         void Inbox(TEventHandlePtr &eh);
-        void Bootstrap(EPriority priority, TAutoPtr<NPageCollection::TFetch>);
         void Dispatch();
         void Handle(ui32 offset, TArrayRef<TLoaded>);
         void Terminate(EStatus code);
 
     private:
-        const TActorId Service;
-        const ui64 Cookie = Max<ui64>();
+        const TActorId StatActorId;
+        const ui64 EventCookie;
         TAutoPtr<NUtil::ILogger> Logger;
 
         /*_ immutable request settings  */
 
-        TActorId Owner;
-        EPriority Priority = EPriority::None;
-        TAutoPtr<NPageCollection::TFetch> Origin;
+        TActorId Sender;
+        EPriority Priority;
+        TIntrusiveConstPtr<NPageCollection::IPageCollection> PageCollection;
+        TVector<TPageId> Pages;
+        NWilson::TTraceId TraceId;
+        ui64 RequestCookie;
 
         /*_ request operational state   */
 
@@ -59,13 +61,10 @@ namespace NBlockIO {
         NMetrics::TTabletIopsRawValue GroupOps;
     };
 
-    template<typename ... TArgs>
-    inline void Start(NActors::IActorOps *ops, TActorId service,
-                        ui64 cookie, TArgs&& ... args)
+    inline void Start(NActors::IActorOps *ops, TActorId statActorId, ui64 cookie, TEvFetch* fetch)
     {
-        auto self = ops->Register(new TBlockIO(service, cookie));
-
-        ops->Send(self, new TEvFetch(std::forward<TArgs>(args)...));
+        auto self = ops->Register(new TBlockIO(statActorId, cookie));
+        ops->Send(self, fetch);
     }
 
 }

ydb/core/tablet_flat/flat_bio_events.h

@@ -1,8 +1,8 @@
 #pragma once
 
 #include "flat_bio_eggs.h"
-#include "flat_sausage_packet.h"
 #include "flat_sausage_fetch.h"
+#include "flat_sausage_gut.h"
 #include <ydb/core/protos/base.pb.h>
 #include <ydb/core/base/events.h>
 #include <ydb/library/actors/core/event_local.h>
@@ -11,6 +11,8 @@ namespace NKikimr {
 namespace NTabletFlatExecutor {
 namespace NBlockIO {
 
+    using TPageId = NPageCollection::TPageId;
+
     enum class EEv : ui32 {
         Base_ = EventSpaceBegin(TKikimrEvents::ES_FLAT_EXECUTOR) + 1088,
 
@@ -20,47 +22,46 @@ namespace NBlockIO {
     };
 
     struct TEvFetch : public TEventLocal<TEvFetch, ui32(EEv::Fetch)> {
-        TEvFetch(EPriority priority, TAutoPtr<NPageCollection::TFetch> fetch)
+        TEvFetch(EPriority priority, TIntrusiveConstPtr<NPageCollection::IPageCollection> pageCollection, TVector<TPageId> pages, ui64 cookie)
             : Priority(priority)
-            , Fetch(fetch)
+            , PageCollection(std::move(pageCollection))
+            , Pages(std::move(pages))
+            , Cookie(cookie)
         {
 
         }
 
-        const EPriority Priority = EPriority::None;
-        TAutoPtr<NPageCollection::TFetch> Fetch;
+        const EPriority Priority;
+        TIntrusiveConstPtr<NPageCollection::IPageCollection> PageCollection;
+        TVector<TPageId> Pages;
+        NWilson::TTraceId TraceId;
+        const ui64 Cookie;
     };
 
     struct TEvData: public TEventLocal<TEvData, ui32(EEv::Data)> {
         using EStatus = NKikimrProto::EReplyStatus;
 
-        TEvData(TAutoPtr<NPageCollection::TFetch> fetch, EStatus status)
+        TEvData(EStatus status, TIntrusiveConstPtr<NPageCollection::IPageCollection> pageCollection, ui64 cookie)
             : Status(status)
-            , Fetch(fetch)
+            , PageCollection(std::move(pageCollection))
+            , Cookie(cookie)
         {
 
         }
 
         void Describe(IOutputStream &out) const
         {
             out
-                << "Blocks{" << Blocks.size() << " pages"
-                << " " << Fetch->PageCollection->Label()
+                << "Blocks{" << Pages.size() << " pages"
+                << " " << PageCollection->Label()
                 << " " << (Status == NKikimrProto::OK ? "ok" : "fail")
                 << " " << NKikimrProto::EReplyStatus_Name(Status) << "}";
         }
 
-        ui64 Bytes() const
-        {
-            return
-                std::accumulate(Blocks.begin(), Blocks.end(), ui64(0),
-                    [](ui64 bytes, const NPageCollection::TLoadedPage& block)
-                        { return bytes + block.Data.size(); });
-        }
-
         const EStatus Status;
-        TAutoPtr<NPageCollection::TFetch> Fetch;
-        TVector<NPageCollection::TLoadedPage> Blocks;
+        TIntrusiveConstPtr<NPageCollection::IPageCollection> PageCollection;
+        TVector<NPageCollection::TLoadedPage> Pages;
+        const ui64 Cookie;
     };
 
 }

ydb/core/tablet_flat/flat_boot_bundle.h

@@ -58,7 +58,8 @@ namespace NBoot {
             LeftReads -= 1;
 
             if (msg.Status == NKikimrProto::OK) {
-                Loader->Save(msg.Cookie, msg.Pages);
+                Y_ENSURE(msg.Cookie == 0);
+                Loader->Save(std::move(msg.Pages));
 
                 TryFinalize();
 
@@ -108,8 +109,8 @@ namespace NBoot {
         void TryFinalize()
         {
             if (!LeftReads) {
-                for (auto req : Loader->Run({.PreloadIndex = true, .PreloadData = false})) {
-                    LeftReads += Logic->LoadPages(this, req);
+                if (auto fetch = Loader->Run({.PreloadIndex = true, .PreloadData = false})) {
+                    LeftReads += Logic->LoadPages(this, std::move(fetch));
                 }
             }