NBS Persistent buffer write to disk (#33263)

Evgenik2 · web-flow · commit 9bfc3dfbf801 · 2026-02-06T16:39:28.000+03:00
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor.cpp b/ydb/core/blobstorage/ddisk/ddisk_actor.cpp
@@ -98,6 +98,7 @@ namespace NKikimr::NDDisk {
             hFunc(NPDisk::TEvChunkReserveResult, Handle)
             hFunc(NPDisk::TEvLogResult, Handle)
             hFunc(TEvPrivate::TEvHandleEventForChunk, Handle)
+            hFunc(TEvPrivate::TEvHandlePersistentBufferEventForChunk, Handle)
             hFunc(NPDisk::TEvCutLog, Handle)
             hFunc(NPDisk::TEvChunkWriteRawResult, Handle)
             hFunc(NPDisk::TEvChunkReadRawResult, Handle)
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor.h b/ydb/core/blobstorage/ddisk/ddisk_actor.h
@@ -3,6 +3,7 @@
 #include "defs.h"
 
 #include "ddisk.h"
+#include "persistent_buffer_space_allocator.h"
 
 #include <ydb/core/blobstorage/vdisk/common/vdisk_config.h>
 #include <ydb/core/blobstorage/pdisk/blobstorage_pdisk.h>
@@ -115,6 +116,7 @@ namespace NKikimr::NDDisk {
             enum {
                 EvHandleSingleQuery = EventSpaceBegin(TEvents::ES_PRIVATE),
                 EvHandleEventForChunk,
+                EvHandlePersistentBufferEventForChunk,
             };
 
             struct TEvHandleEventForChunk : TEventLocal<TEvHandleEventForChunk, EvHandleEventForChunk> {
@@ -126,6 +128,14 @@ namespace NKikimr::NDDisk {
                     , VChunkIndex(vChunkIndex)
                 {}
             };
+
+            struct TEvHandlePersistentBufferEventForChunk : TEventLocal<TEvHandlePersistentBufferEventForChunk, EvHandlePersistentBufferEventForChunk> {
+                ui32 ChunkIndex;
+
+                TEvHandlePersistentBufferEventForChunk(ui32 chunkIndex)
+                    : ChunkIndex(chunkIndex)
+                {}
+            };
         };
 
     public:
@@ -200,14 +210,23 @@ namespace NKikimr::NDDisk {
         std::queue<std::variant<TChunkForData, TChunkForPersistentBuffer>> ChunkAllocateQueue;
         THashMap<ui64, std::function<void()>> LogCallbacks;
         ui64 NextCookie = 1;
-        THashMap<ui64, std::tuple<NWilson::TSpan, std::function<void(NPDisk::TEvChunkWriteRawResult&, NWilson::TSpan&&)>>> WriteCallbacks;
+
+        struct TPendingWrite {
+            NWilson::TSpan Span;
+            std::function<void(NPDisk::TEvChunkWriteRawResult&, NWilson::TSpan&&)> Callback;
+        };
+
+        using TPersistentBufferPendingWrite = std::function<void(NPDisk::TEvChunkWriteRawResult&)>;
+
+        THashMap<ui64, std::variant<TPendingWrite, TPersistentBufferPendingWrite>> WriteCallbacks;
         THashMap<ui64, std::tuple<NWilson::TSpan, std::function<void(NPDisk::TEvChunkReadRawResult&, NWilson::TSpan&&)>>> ReadCallbacks;
 
         void IssueChunkAllocation(ui64 tabletId, ui64 vChunkIndex);
         void Handle(NPDisk::TEvChunkReserveResult::TPtr ev);
         void HandleChunkReserved();
         void Handle(NPDisk::TEvLogResult::TPtr ev);
         void Handle(TEvPrivate::TEvHandleEventForChunk::TPtr ev);
+        void Handle(TEvPrivate::TEvHandlePersistentBufferEventForChunk::TPtr ev);
 
         void Handle(NPDisk::TEvCutLog::TPtr ev);
 
@@ -312,6 +331,7 @@ namespace NKikimr::NDDisk {
             struct TRecord {
                 ui32 OffsetInBytes;
                 ui32 Size;
+                std::vector<TPersistentBufferSectorInfo> Sectors;
                 TRope Data;
             };
 
@@ -320,10 +340,50 @@ namespace NKikimr::NDDisk {
 
         std::map<std::tuple<ui64, ui64>, TPersistentBuffer> PersistentBuffers;
 
-        std::set<TChunkIdx> PersistentBufferOwnedChunks;
+        static constexpr ui32 SectorSize = 4096;
+        static constexpr ui32 SectorInChunk = 32768;
+        static constexpr ui32 ChunkSize = SectorSize * SectorInChunk;
+        static constexpr ui32 MaxChunks = 128;
+        static constexpr ui32 MaxSectorsPerBuffer = 128;
+
+
+        struct TPersistentBufferHeader {
+            static constexpr ui8 PersistentBufferHeaderSignature[16] = {249, 173, 163, 160, 196, 193, 69, 133, 83, 38, 34, 104, 170, 146, 237, 156};
+            static constexpr ui32 HeaderChecksumOffset = 24;
+            static constexpr ui32 HeaderChecksumSize = 8;
+            ui8 Signature[16];
+            ui64 HeaderChecksum;
+            ui64 TabletId;
+            ui64 VChunkIndex;
+            ui32 OffsetInBytes;
+            ui32 Size;
+            ui64 Lsn;
+            TPersistentBufferSectorInfo Locations[MaxSectorsPerBuffer];
+        };
+        static_assert(sizeof(TPersistentBufferHeader) <= ChunkSize);
+
+        bool IssuePersistentBufferChunkAllocationInflight = false;
+        struct TPersistentBufferToDiskWriteInFlight {
+            TActorId Sender;
+            ui64 Cookie;
+            TActorId Session;
+            ui32 OffsetInBytes;
+            ui32 Size;
+            std::set<ui64> WriteCookies;
+            std::vector<TPersistentBufferSectorInfo> Sectors;
+            TRope Data;
+            NWilson::TSpan Span;
+        };
+        std::map<std::tuple<ui64, ui64, ui64>, TPersistentBufferToDiskWriteInFlight> PersistentBufferWriteInflight;
+
+        TPersistentBufferSpaceAllocator PersistentBufferSpaceAllocator;
+
         ui64 PersistentBufferChunkMapSnapshotLsn = Max<ui64>();
+        std::queue<TPendingEvent> PendingPersistentBufferEvents;
 
         void IssuePersistentBufferChunkAllocation();
+        void ProcessPersistentBufferQueue();
+        std::vector<std::tuple<ui32, ui32, TRope>> SlicePersistentBuffer(ui64 tabletId, ui64 vchunkIndex, ui64 lsn, ui32 offsetInBytes, ui32 size, TRope&& data, const std::vector<TPersistentBufferSectorInfo>& sectors);
 
         struct TWriteInFlight {
             TActorId Sender;
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor_boot.cpp b/ydb/core/blobstorage/ddisk/ddisk_actor_boot.cpp
@@ -44,7 +44,7 @@ namespace NKikimr::NDDisk {
             const bool success = chunkMap.ParseFromArray(record.Data.data(), record.Data.size());
             Y_ABORT_UNLESS(success);
             for (auto idx : chunkMap.GetChunkIdxs()) {
-                PersistentBufferOwnedChunks.insert(idx);
+                PersistentBufferSpaceAllocator.AddNewChunk(idx);
                 ++*Counters.Chunks.ChunksOwned;
             }
         }
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor_chunks.cpp b/ydb/core/blobstorage/ddisk/ddisk_actor_chunks.cpp
@@ -11,8 +11,11 @@ namespace NKikimr::NDDisk {
     }
 
     void TDDiskActor::IssuePersistentBufferChunkAllocation() {
-        ChunkAllocateQueue.emplace(TChunkForPersistentBuffer{});
-        HandleChunkReserved();
+        if (!IssuePersistentBufferChunkAllocationInflight) {
+            IssuePersistentBufferChunkAllocationInflight = true;
+            ChunkAllocateQueue.emplace(TChunkForPersistentBuffer{});
+            HandleChunkReserved();
+        }
     }
 
     void TDDiskActor::Handle(NPDisk::TEvChunkReserveResult::TPtr ev) {
@@ -67,11 +70,10 @@ namespace NKikimr::NDDisk {
                     ChunkMapIncrementsInFlight.emplace(tabletId, vChunkIndex, chunkIdx);
                 },
                 [this, chunkIdx](const TChunkForPersistentBuffer&) {
-                    Y_ABORT_UNLESS(!PersistentBufferOwnedChunks.contains(chunkIdx));
                     IssuePDiskLogRecord(TLogSignature::SignaturePersistentBufferChunkMap, chunkIdx
                         , CreatePersistentBufferChunkMapSnapshot(), &PersistentBufferChunkMapSnapshotLsn, [this, chunkIdx] {
-                        PersistentBufferOwnedChunks.insert(chunkIdx);
-                        // TODO: Send(SelfId(), new TEvPrivate::TEvHandlePersistentBufferEventForChunk(chunkIdx));
+                        PersistentBufferSpaceAllocator.AddNewChunk(chunkIdx);
+                        Send(SelfId(), new TEvPrivate::TEvHandlePersistentBufferEventForChunk(chunkIdx));
                         ++*Counters.Chunks.ChunksOwned;
                     });
                 }
@@ -123,7 +125,7 @@ namespace NKikimr::NDDisk {
 
     NKikimrBlobStorage::NDDisk::NInternal::TPersistentBufferChunkMapLogRecord TDDiskActor::CreatePersistentBufferChunkMapSnapshot() {
         NKikimrBlobStorage::NDDisk::NInternal::TPersistentBufferChunkMapLogRecord record;
-        for (const auto& chunkIdx : PersistentBufferOwnedChunks) {
+        for (const auto& chunkIdx : PersistentBufferSpaceAllocator.OwnedChunks) {
             record.AddChunkIdxs(chunkIdx);
         }
         return record;
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor_persistent_buffer.cpp b/ydb/core/blobstorage/ddisk/ddisk_actor_persistent_buffer.cpp
@@ -1,55 +1,191 @@
 #include "ddisk_actor.h"
 
 #include <ydb/core/util/stlog.h>
-
 #include <ydb/core/util/pb.h>
 
+#define XXH_INLINE_ALL
+#include <contrib/libs/xxhash/xxhash.h>
+
 namespace NKikimr::NDDisk {
 
-    void TDDiskActor::Handle(TEvWritePersistentBuffer::TPtr ev) {
-        if (!CheckQuery(*ev, &Counters.Interface.WritePersistentBuffer)) {
-            return;
+    void TDDiskActor::Handle(TEvPrivate::TEvHandlePersistentBufferEventForChunk::TPtr ev) {
+        auto chunkIdx = ev->Get()->ChunkIndex;
+        Y_ABORT_UNLESS(chunkIdx);
+        ProcessPersistentBufferQueue();
+    }
+
+    ui64 CalculateChecksum(const TRope::TIterator begin, size_t numBytes) {
+        XXH3_state_t state;
+        XXH3_64bits_reset(&state);
+
+        for (auto it = begin; numBytes && it.Valid(); it.AdvanceToNextContiguousBlock()) {
+            const size_t n = Min(numBytes, it.ContiguousSize());
+            XXH3_64bits_update(&state, it.ContiguousData(), n);
+            numBytes -= n;
         }
 
-        const auto& record = ev->Get()->Record;
+        return XXH3_64bits_digest(&state);
+    }
+
+    std::vector<std::tuple<ui32, ui32, TRope>> TDDiskActor::SlicePersistentBuffer(ui64 tabletId, ui64 vchunkIndex,
+        ui64 lsn, ui32 offsetInBytes, ui32 sizeInBytes, TRope&& payload, const std::vector<TPersistentBufferSectorInfo>& sectors) {
+        auto headerData = TRcBuf::Uninitialized(SectorSize);
+        TPersistentBufferHeader *header = (TPersistentBufferHeader*)headerData.GetDataMut();
+        memcpy(header->Signature, TPersistentBufferHeader::PersistentBufferHeaderSignature, 16);
+        header->TabletId = tabletId;
+        header->VChunkIndex = vchunkIndex;
+        header->OffsetInBytes = offsetInBytes;
+        header->Size = sizeInBytes;
+        header->Lsn = lsn;
+
+        for (ui32 i = 1; i < sectors.size(); ++i) {
+            auto& loc = header->Locations[i - 1];
+            loc = sectors[i];
+            auto it = payload.Position(SectorSize * (i - 1));
+            if (memcmp((*it).first, header->Signature, 16) != 0) {
+                loc.HasSignatureCorrection = true;
+                *payload.Position(SectorSize * (i - 1)).ContiguousDataMut() = 0;
+            }
+            loc.Checksum = CalculateChecksum(payload.Position(SectorSize * (i - 1)), SectorSize);
+        }
+        header->HeaderChecksum = 0;
+        std::vector<std::tuple<ui32, ui32, TRope>> parts;
+        parts.reserve(sectors.size());
+        for (ui32 sectorIdx = 0, first = 0; sectorIdx <= sectors.size(); sectorIdx++) {
+            if (sectorIdx == sectors.size()
+                || sectors[first].ChunkIdx != sectors[sectorIdx].ChunkIdx
+                || sectors[first].SectorIdx != sectors[sectorIdx].SectorIdx + sectorIdx - first) {
+                TRope data;
+                ui32 partSize = (sectorIdx - (first == 0 ? 1 : first)) * SectorSize;
+                if (first == 0) {
+                    data = headerData;
+                    auto cs = CalculateChecksum(data.Position(TPersistentBufferHeader::HeaderChecksumOffset + TPersistentBufferHeader::HeaderChecksumSize)
+                        , SectorSize - TPersistentBufferHeader::HeaderChecksumOffset - TPersistentBufferHeader::HeaderChecksumSize);
+                    memcpy(data.Position(TPersistentBufferHeader::HeaderChecksumOffset).ContiguousDataMut(), &cs, TPersistentBufferHeader::HeaderChecksumSize);
+
+                }
+                payload.ExtractFront(partSize, &data);
+                parts.emplace_back(sectors[first].ChunkIdx, sectors[first].SectorIdx * SectorSize, std::move(data));
+                first = sectorIdx;
+            }
+        }
+        return parts;
+    }
+
+    void TDDiskActor::ProcessPersistentBufferQueue() {
+        Y_ABORT_UNLESS(!PendingPersistentBufferEvents.empty());
+        auto& temp = PendingPersistentBufferEvents.front().Ev;
+        const auto& record = temp->Get<TEvWritePersistentBuffer>()->Record;
         const TQueryCredentials creds(record.GetCredentials());
         const TBlockSelector selector(record.GetSelector());
         const ui64 lsn = record.GetLsn();
+        ui32 sectorsCnt = selector.Size / SectorSize;
+        const auto sectors = PersistentBufferSpaceAllocator.Occupy(sectorsCnt);
+        if (sectors.size() == 0) {
+            IssuePersistentBufferChunkAllocation();
+            return;
+        }
+        Y_ABORT_UNLESS(sectors.size() == sectorsCnt && sectorsCnt <= MaxSectorsPerBuffer);
 
-        Counters.Interface.WritePersistentBuffer.Request(selector.Size);
-
-        auto span = std::move(NWilson::TSpan(TWilson::DDiskTopLevel, std::move(ev->TraceId), "DDisk.WritePersistentBuffer",
+        const TWriteInstruction instr(record.GetInstruction());
+        TRope payload;
+        if (instr.PayloadId) {
+            payload = temp->Get<TEvWritePersistentBuffer>()->GetPayload(*instr.PayloadId);
+        }
+        auto span = std::move(NWilson::TSpan(TWilson::DDiskTopLevel, std::move(temp->TraceId), "DDisk.WritePersistentBuffer",
                 NWilson::EFlags::NONE, TActivationContext::ActorSystem())
             .Attribute("tablet_id", static_cast<long>(creds.TabletId))
             .Attribute("vchunk_index", static_cast<long>(selector.VChunkIndex))
             .Attribute("offset_in_bytes", selector.OffsetInBytes)
             .Attribute("size", selector.Size)
             .Attribute("lsn", static_cast<long>(lsn)));
+        Counters.Interface.WritePersistentBuffer.Request(selector.Size);
 
-        const TWriteInstruction instr(record.GetInstruction());
-        TRope data;
-        if (instr.PayloadId) {
-            data = ev->Get()->GetPayload(*instr.PayloadId);
+        auto parts = SlicePersistentBuffer(creds.TabletId,
+            selector.VChunkIndex, lsn, selector.OffsetInBytes, selector.Size, TRope(payload), sectors);
+
+        auto& inflightRecord = PersistentBufferWriteInflight[{creds.TabletId, selector.VChunkIndex, lsn}];
+        inflightRecord = {
+            .Sender = temp->Sender,
+            .Cookie = temp->Cookie,
+            .Session = temp->InterconnectSession,
+            .OffsetInBytes = selector.OffsetInBytes,
+            .Size = selector.Size,
+            .Sectors = std::move(sectors),
+            .Data = std::move(payload),
+            .Span = std::move(span),
+        };
+
+        for(auto& [chunkIdx, offset, data] : parts) {
+            const ui64 cookie = NextCookie++;
+            inflightRecord.WriteCookies.insert(cookie);
+
+            Send(BaseInfo.PDiskActorID, new NPDisk::TEvChunkWriteRaw(
+                PDiskParams->Owner,
+                PDiskParams->OwnerRound,
+                chunkIdx,
+                offset,
+                std::move(data)), 0, cookie);
+
+            WriteCallbacks.try_emplace(cookie, [this, writeCookie = cookie, tabletId = creds.TabletId,
+                vchunkIndex = selector.VChunkIndex, lsn = lsn](NPDisk::TEvChunkWriteRawResult& /*ev*/) {
+                auto itInflight = PersistentBufferWriteInflight.find({tabletId, vchunkIndex, lsn});
+                Y_ABORT_UNLESS(itInflight != PersistentBufferWriteInflight.end());
+                auto& inflight = itInflight->second;
+                auto eraseCnt = inflight.WriteCookies.erase(writeCookie);
+                Y_ABORT_UNLESS(eraseCnt == 1);
+                if (inflight.WriteCookies.empty()) {
+                    Counters.Interface.WritePersistentBuffer.Reply(true);
+                    inflight.Span.End();
+                    auto& buffer = PersistentBuffers[{tabletId, vchunkIndex}];
+                    auto [it, inserted] = buffer.Records.try_emplace(lsn);
+                    TPersistentBuffer::TRecord& pr = it->second;
+                    if (inserted) {
+                        pr = {
+                            .OffsetInBytes = inflight.OffsetInBytes,
+                            .Size = inflight.Size,
+                            .Sectors = std::move(inflight.Sectors),
+                            .Data = std::move(inflight.Data),
+                        };
+                    } else {
+                        Y_ABORT_UNLESS(pr.OffsetInBytes == inflight.OffsetInBytes);
+                        Y_ABORT_UNLESS(pr.Size == inflight.Size);
+                        Y_ABORT_UNLESS(pr.Data == inflight.Data);
+                    }
+                    auto replyEv = std::make_unique<TEvWritePersistentBufferResult>(NKikimrBlobStorage::NDDisk::TReplyStatus::OK);
+                    auto h = std::make_unique<IEventHandle>(inflight.Sender, SelfId(), replyEv.release(), 0, inflight.Cookie);
+                    if (inflight.Session) {
+                        h->Rewrite(TEvInterconnect::EvForward, inflight.Session);
+                    }
+                    TActivationContext::Send(h.release());
+                    PersistentBufferWriteInflight.erase(itInflight);
+                }
+            });
         }
 
-        auto& buffer = PersistentBuffers[{creds.TabletId, selector.VChunkIndex}];
-        auto [it, inserted] = buffer.Records.try_emplace(lsn);
-        TPersistentBuffer::TRecord& pr = it->second;
-        if (inserted) {
-            pr = {
-                .OffsetInBytes = selector.OffsetInBytes,
-                .Size = selector.Size,
-                .Data = std::move(data),
-            };
-        } else {
-            Y_ABORT_UNLESS(pr.OffsetInBytes == selector.OffsetInBytes);
-            Y_ABORT_UNLESS(pr.Size == selector.Size);
-            Y_ABORT_UNLESS(pr.Data == data);
+        PendingPersistentBufferEvents.pop();
+        if (!PendingPersistentBufferEvents.empty()) {
+            ProcessPersistentBufferQueue();
         }
+    }
 
-        Counters.Interface.WritePersistentBuffer.Reply(true);
-        span.End();
-        SendReply(*ev, std::make_unique<TEvWritePersistentBufferResult>(NKikimrBlobStorage::NDDisk::TReplyStatus::OK));
+    void TDDiskActor::Handle(TEvWritePersistentBuffer::TPtr ev) {
+        if (!CheckQuery(*ev, &Counters.Interface.WritePersistentBuffer)) {
+            return;
+        }
+        const auto& record = ev->Get()->Record;
+        const TBlockSelector selector(record.GetSelector());
+        if (selector.Size > MaxSectorsPerBuffer * SectorSize) {
+            Counters.Interface.WritePersistentBuffer.Request(selector.Size);
+            Counters.Interface.WritePersistentBuffer.Reply(false);
+            SendReply(*ev, std::make_unique<TEvWritePersistentBufferResult>(
+                NKikimrBlobStorage::NDDisk::TReplyStatus::INCORRECT_REQUEST,
+                TStringBuilder() << "persistent buffer write limit "
+                    << (MaxSectorsPerBuffer * SectorSize) << " bytes, received " << selector.Size << " bytes"));
+            return;
+        }
+        PendingPersistentBufferEvents.emplace(ev, "WaitingPersistentBufferWrite");
+        ProcessPersistentBufferQueue();
     }
 
     void TDDiskActor::Handle(TEvReadPersistentBuffer::TPtr ev) {
@@ -200,6 +336,8 @@ namespace NKikimr::NDDisk {
         Y_ABORT_UNLESS(pr.OffsetInBytes == selector.OffsetInBytes);
         Y_ABORT_UNLESS(pr.Size == selector.Size);
 
+        PersistentBufferSpaceAllocator.Free(pr.Sectors);
+
         buffer.Records.erase(jt);
         if (buffer.Records.empty()) {
             PersistentBuffers.erase(it);
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor_read_write.cpp b/ydb/core/blobstorage/ddisk/ddisk_actor_read_write.cpp

Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ namespace NKikimr::NDDisk {`
`44`	`44`	`const bool success = chunkMap.ParseFromArray(record.Data.data(), record.Data.size());`
`45`	`45`	`Y_ABORT_UNLESS(success);`
`46`	`46`	`for (auto idx : chunkMap.GetChunkIdxs()) {`
`47`		`- PersistentBufferOwnedChunks.insert(idx);`
	`47`	`+ PersistentBufferSpaceAllocator.AddNewChunk(idx);`
`48`	`48`	`++*Counters.Chunks.ChunksOwned;`
`49`	`49`	`}`
`50`	`50`	`}`