issue-539: add tablet-side two-stage writes support for scaling throughput

cloud/filestore/config/storage.proto

@@ -216,4 +216,12 @@ message TStorageConfig
     optional uint32 EntryTimeout = 335;
     optional uint32 NegativeEntryTimeout = 336;
     optional uint32 AttrTimeout = 337;
+
+    // Enables GenerateBlobs + WriteBlob + AddData instead of WriteBlob
+    // for writing.
+    optional bool TwoStageWriteEnabled = 338;
+
+    // When issuing blobs, the tablet acquires collect barrier. In order to
+    // release it in case of a client disconnect, this timeout is used.
+    optional uint32 GenerateBlobsReleaseCollectBarrierTimeout = 339;
 }

cloud/filestore/libs/storage/api/tablet.h

@@ -24,6 +24,8 @@ namespace NCloud::NFileStore::NStorage {
     xxx(ChangeStorageConfig,        __VA_ARGS__)                               \
     xxx(DescribeData,               __VA_ARGS__)                               \
     xxx(DescribeSessions,           __VA_ARGS__)                               \
+    xxx(GenerateBlobs,              __VA_ARGS__)                               \
+    xxx(AddData,                    __VA_ARGS__)                               \
 // FILESTORE_TABLET_REQUESTS
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -65,6 +67,12 @@ struct TEvIndexTablet
         EvDescribeSessionsRequest = EvBegin + 17,
         EvDescribeSessionsResponse,
 
+        EvGenerateBlobsRequest = EvBegin + 19,
+        EvGenerateBlobsResponse,
+
+        EvAddDataRequest = EvBegin + 21,
+        EvAddDataResponse,
+
         EvEnd
     };
 

cloud/filestore/libs/storage/core/config.cpp

@@ -133,11 +133,16 @@ namespace {
             NCloud::NProto::AUTHORIZATION_IGNORE                              )\
                                                                                \
     xxx(TwoStageReadEnabled,             bool,      false                     )\
+    xxx(TwoStageWriteEnabled,            bool,      false                     )\
     xxx(EntryTimeout,                    TDuration, TDuration::Zero()         )\
     xxx(NegativeEntryTimeout,            TDuration, TDuration::Zero()         )\
     xxx(AttrTimeout,                     TDuration, TDuration::Zero()         )\
     xxx(MaxOutOfOrderCompactionMapLoadRequestsInQueue,  ui32,      5          )\
     xxx(MaxBackpressureErrorsBeforeSuicide,             ui32,      1000       )\
+                                                                               \
+    xxx(GenerateBlobsReleaseCollectBarrierTimeout,                                 \
+        TDuration,                                                             \
+        TDuration::Seconds(10)                                                )\
 // FILESTORE_STORAGE_CONFIG
 
 #define FILESTORE_DECLARE_CONFIG(name, type, value)                            \

cloud/filestore/libs/storage/core/config.h

@@ -175,6 +175,7 @@ class TStorageConfig
     NCloud::NProto::EAuthorizationMode GetAuthorizationMode() const;
 
     bool GetTwoStageReadEnabled() const;
+    bool GetTwoStageWriteEnabled() const;
     TDuration GetEntryTimeout() const;
     TDuration GetNegativeEntryTimeout() const;
     TDuration GetAttrTimeout() const;
@@ -185,6 +186,8 @@ class TStorageConfig
 
     ui32 GetMaxBackpressureErrorsBeforeSuicide() const;
 
+    TDuration GetGenerateBlobsReleaseCollectBarrierTimeout() const;
+
     void Dump(IOutputStream& out) const;
     void DumpHtml(IOutputStream& out) const;
     void DumpOverridesHtml(IOutputStream& out) const;

cloud/filestore/libs/storage/model/block_buffer.cpp

@@ -44,6 +44,11 @@ class TBlockBuffer final
         return { ptr, ByteRange.BlockSize };
     }
 
+    ui32 GetBlockSize() const override
+    {
+        return ByteRange.BlockSize;
+    }
+
     void SetBlock(size_t index, TStringBuf block) override
     {
         Y_ABORT_UNLESS(block.size() == ByteRange.BlockSize);
@@ -107,6 +112,11 @@ class TLazyBlockBuffer final
         return block;
     }
 
+    ui32 GetBlockSize() const override
+    {
+        return ByteRange.BlockSize;
+    }
+
     void SetBlock(size_t index, TStringBuf block) override
     {
         Y_ABORT_UNLESS(block.size() == ByteRange.BlockSize);

cloud/filestore/libs/storage/model/block_buffer.h

@@ -17,6 +17,7 @@ struct IBlockBuffer
 
     virtual TStringBuf GetUnalignedHead() = 0;
     virtual TStringBuf GetBlock(size_t index) = 0;
+    virtual ui32 GetBlockSize() const = 0;
     virtual TStringBuf GetUnalignedTail() = 0;
     virtual void SetBlock(size_t index, TStringBuf block) = 0;
     virtual void ClearBlock(size_t index) = 0;

cloud/filestore/libs/storage/tablet/model/blob_builder.cpp

@@ -70,16 +70,18 @@ void TMergedBlobBuilder::Accept(
     const TBlock& block,
     ui32 blocksCount,
     ui32 blockOffset,
-    IBlockBuffer& buffer)
+    IBlockBuffer& buffer,
+    bool shouldFillBlobContent)
 {
     TString blobContent;
     blobContent.reserve(blocksCount * BlockSize);
 
     for (size_t i = 0; i < blocksCount; ++i) {
-        auto blockData = buffer.GetBlock(blockOffset + i);
-        Y_ABORT_UNLESS(blockData.size() == BlockSize);
-
-        blobContent.append(blockData);
+        Y_ABORT_UNLESS(buffer.GetBlockSize() == BlockSize);
+        if (shouldFillBlobContent) {
+            auto blockData = buffer.GetBlock(blockOffset + i);
+            blobContent.append(blockData);
+        }
     }
 
     BlocksCount += blocksCount;

cloud/filestore/libs/storage/tablet/model/blob_builder.h

@@ -91,7 +91,8 @@ class TMergedBlobBuilder
         const TBlock& block,
         ui32 blocksCount,
         ui32 blockOffset,
-        IBlockBuffer& buffer);
+        IBlockBuffer& buffer,
+        bool shouldFillBlobContent);
 
     TVector<TMergedBlob> Finish();
 

cloud/filestore/libs/storage/tablet/model/garbage_queue.cpp

@@ -182,11 +182,14 @@ void TGarbageQueue::AcquireCollectBarrier(ui64 commitId)
     }
 }
 
-void TGarbageQueue::ReleaseCollectBarrier(ui64 commitId)
+void TGarbageQueue::ReleaseCollectBarrier(ui64 commitId, bool allowMissing)
 {
     {
         auto it = Impl->Barriers.find(commitId);
-        Y_ABORT_UNLESS(it != Impl->Barriers.end());
+        Y_ABORT_UNLESS(it != Impl->Barriers.end() || allowMissing);
+        if (it == Impl->Barriers.end()) {
+            return;
+        }
 
         auto& barrier = const_cast<TBarrier&>(*it);
 
@@ -210,6 +213,12 @@ void TGarbageQueue::ReleaseCollectBarrier(ui64 commitId)
     }
 }
 
+bool TGarbageQueue::IsCollectBarrierAcquired(ui64 commitId) const
+{
+    auto it = Impl->Barriers.find(commitId);
+    return it != Impl->Barriers.end() && it->RefCount > 0;
+}
+
 ui64 TGarbageQueue::GetCollectCommitId() const
 {
     if (Impl->Barriers) {

cloud/filestore/libs/storage/tablet/model/garbage_queue.h

@@ -46,7 +46,12 @@ class TGarbageQueue
     //
 
     void AcquireCollectBarrier(ui64 commitId);
-    void ReleaseCollectBarrier(ui64 commitId);
+    /**
+     * @param allowMissing If true, there will not be an abort if the barrier is
+     * missing.
+     */
+    void ReleaseCollectBarrier(ui64 commitId, bool allowMissing);
+    bool IsCollectBarrierAcquired(ui64 commitId) const;
 
     ui64 GetCollectCommitId() const;
 };

cloud/filestore/libs/storage/tablet/tablet_actor.cpp

@@ -492,6 +492,7 @@ STFUNC(TIndexTabletActor::StateBoot)
         IgnoreFunc(TEvLocal::TEvTabletMetrics);
         IgnoreFunc(TEvIndexTabletPrivate::TEvUpdateCounters);
         IgnoreFunc(TEvIndexTabletPrivate::TEvUpdateLeakyBucketCounters);
+        IgnoreFunc(TEvIndexTabletPrivate::TEvReleaseCollectBarrier);
 
         IgnoreFunc(TEvHiveProxy::TEvReassignTabletResponse);
 
@@ -515,6 +516,7 @@ STFUNC(TIndexTabletActor::StateInit)
         HFunc(TEvFileStore::TEvUpdateConfig, HandleUpdateConfig);
         HFunc(TEvIndexTabletPrivate::TEvUpdateCounters, HandleUpdateCounters);
         HFunc(TEvIndexTabletPrivate::TEvUpdateLeakyBucketCounters, HandleUpdateLeakyBucketCounters);
+        HFunc(TEvIndexTabletPrivate::TEvReleaseCollectBarrier, HandleReleaseCollectBarrier);
 
         FILESTORE_HANDLE_REQUEST(WaitReady, TEvIndexTablet)
 
@@ -542,6 +544,8 @@ STFUNC(TIndexTabletActor::StateWork)
         HFunc(TEvIndexTabletPrivate::TEvUpdateCounters, HandleUpdateCounters);
         HFunc(TEvIndexTabletPrivate::TEvUpdateLeakyBucketCounters, HandleUpdateLeakyBucketCounters);
 
+        HFunc(TEvIndexTabletPrivate::TEvReleaseCollectBarrier, HandleReleaseCollectBarrier);
+
         HFunc(TEvents::TEvWakeup, HandleWakeup);
         HFunc(TEvents::TEvPoisonPill, HandlePoisonPill);
 
@@ -579,6 +583,8 @@ STFUNC(TIndexTabletActor::StateZombie)
         IgnoreFunc(TEvIndexTabletPrivate::TEvUpdateCounters);
         IgnoreFunc(TEvIndexTabletPrivate::TEvUpdateLeakyBucketCounters);
 
+        IgnoreFunc(TEvIndexTabletPrivate::TEvReleaseCollectBarrier);
+
         IgnoreFunc(TEvIndexTabletPrivate::TEvReadDataCompleted);
         IgnoreFunc(TEvIndexTabletPrivate::TEvWriteDataCompleted);
 
@@ -604,6 +610,7 @@ STFUNC(TIndexTabletActor::StateBroken)
     switch (ev->GetTypeRewrite()) {
         IgnoreFunc(TEvIndexTabletPrivate::TEvUpdateCounters);
         IgnoreFunc(TEvIndexTabletPrivate::TEvUpdateLeakyBucketCounters);
+        IgnoreFunc(TEvIndexTabletPrivate::TEvReleaseCollectBarrier);
 
         HFunc(TEvents::TEvPoisonPill, HandlePoisonPill);
         HFunc(TEvTablet::TEvTabletDead, HandleTabletDead);

cloud/filestore/libs/storage/tablet/tablet_actor.h

@@ -391,6 +391,10 @@ class TIndexTabletActor final
         const TEvIndexTabletPrivate::TEvUpdateLeakyBucketCounters::TPtr& ev,
         const NActors::TActorContext& ctx);
 
+    void HandleReleaseCollectBarrier(
+        const TEvIndexTabletPrivate::TEvReleaseCollectBarrier::TPtr& ev,
+        const NActors::TActorContext& ctx);
+
     void HandleReadDataCompleted(
         const TEvIndexTabletPrivate::TEvReadDataCompleted::TPtr& ev,
         const NActors::TActorContext& ctx);