24-4 Allow to send duplicates through TDedicatedPipePool (#18148)

Author: kungasc

ydb/core/tx/schemeshard/dedicated_pipe_pool.h

@@ -17,18 +17,21 @@ class TDedicatedPipePool {
     TMap<TActorId, std::pair<TEntityId, TTabletId>> Owners;
 
 public:
-    void Create(const TEntityId& entityId, TTabletId dst, THolder<IEventBase> message, const TActorContext& ctx) {
-        Y_ABORT_UNLESS(!Pipes[entityId].contains(dst));
+    void Send(const TEntityId& entityId, TTabletId dst, THolder<IEventBase> message, const TActorContext& ctx) {
         using namespace NTabletPipe;
 
-        const auto clientId = ctx.ExecutorThread.RegisterActor(CreateClient(ctx.SelfID, ui64(dst), TClientRetryPolicy {
-            .MinRetryTime = TDuration::MilliSeconds(100),
-            .MaxRetryTime = TDuration::Seconds(30),
-        }));
+        if (!Pipes[entityId].contains(dst)) {
+            const auto clientId = ctx.ExecutorThread.RegisterActor(CreateClient(ctx.SelfID, ui64(dst), TClientRetryPolicy {
+                .MinRetryTime = TDuration::MilliSeconds(100),
+                .MaxRetryTime = TDuration::Seconds(30),
+            }));
 
-        Pipes[entityId][dst] = clientId;
-        Owners[clientId] = std::make_pair(entityId, dst);
+            Pipes[entityId][dst] = clientId;
+            Owners[clientId] = std::make_pair(entityId, dst);
+        }
 
+        const auto clientId = Pipes[entityId][dst];
+        Y_ABORT_UNLESS(Owners[clientId] == std::make_pair(entityId, dst));
         SendData(ctx.SelfID, clientId, message.Release(), 0);
     }
 
@@ -55,10 +58,10 @@ class TDedicatedPipePool {
         }
     }
 
-    ui64 CloseAll(const TEntityId& entityId, const TActorContext& ctx) {
+    void CloseAll(const TEntityId& entityId, const TActorContext& ctx) {
         auto entityIt = Pipes.find(entityId);
         if (entityIt == Pipes.end()) {
-            return 0;
+            return;
         }
 
         const auto& entityPipes = entityIt->second;
@@ -71,8 +74,6 @@ class TDedicatedPipePool {
         for (const auto& tabletId : tablets) {
             Close(entityId, tabletId, ctx);
         }
-
-        return tablets.size();
     }
 
     void Shutdown(const TActorContext& ctx) {

ydb/core/tx/schemeshard/schemeshard_build_index__progress.cpp

@@ -194,7 +194,7 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
 private:
     TIndexBuildId BuildId;
 
-    TDeque<std::tuple<TTabletId, ui64, THolder<IEventBase>>> ToTabletSend;
+    TMap<TTabletId, THolder<IEventBase>> ToTabletSend;
 
 public:
     explicit TTxProgress(TSelf* self, TIndexBuildId id)
@@ -206,10 +206,8 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
         Y_ABORT_UNLESS(Self->IndexBuilds.contains(BuildId));
         TIndexBuildInfo::TPtr buildInfo = Self->IndexBuilds.at(BuildId);
 
-        LOG_I("TTxBuildProgress: Resume"
-              << ": id# " << BuildId);
-        LOG_D("TTxBuildProgress: Resume"
-              << ": " << *buildInfo);
+        LOG_N("TTxBuildProgress: Execute: " << BuildId << " " << buildInfo->State);
+        LOG_D("TTxBuildProgress: Execute: " << BuildId << " " << buildInfo->State << " " << *buildInfo);
 
         switch (buildInfo->State) {
         case TIndexBuildInfo::EState::Invalid:
@@ -263,6 +261,7 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
                 buildInfo->DoneShards.clear();
                 buildInfo->InProgressShards.clear();
 
+                ToTabletSend.clear();
                 Self->IndexBuildPipes.CloseAll(BuildId, ctx);
 
                 ChangeState(BuildId, TIndexBuildInfo::EState::Cancellation_Applying);
@@ -283,6 +282,9 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
                 && buildInfo->DoneShards.empty()
                 && buildInfo->InProgressShards.empty())
             {
+                ToTabletSend.clear();
+                Self->IndexBuildPipes.CloseAll(BuildId, ctx);
+
                 for (const auto& item: buildInfo->Shards) {
                     const TIndexBuildInfo::TShardStatus& shardStatus = item.second;
                     switch (shardStatus.Status) {
@@ -374,16 +376,18 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
                 ev->Record.SetSeqNoGeneration(Self->Generation());
                 ev->Record.SetSeqNoRound(++shardStatus.SeqNoRound);
 
-                LOG_D("TTxBuildProgress: TEvBuildIndexCreateRequest"
+                LOG_N("TTxBuildProgress: TEvBuildIndexCreateRequest"
                       << ": " << ev->Record.ShortDebugString());
 
-                ToTabletSend.emplace_back(shardId, ui64(BuildId), std::move(ev));
+                ToTabletSend.emplace(shardId, std::move(ev));
             }
 
             if (buildInfo->InProgressShards.empty() && buildInfo->ToUploadShards.empty()
-                && buildInfo->DoneShards.size() == buildInfo->Shards.size()) {
+                && buildInfo->DoneShards.size() == buildInfo->Shards.size())
+            {
                 // all done
-                Y_ABORT_UNLESS(0 == Self->IndexBuildPipes.CloseAll(BuildId, ctx));
+                ToTabletSend.clear();
+                Self->IndexBuildPipes.CloseAll(BuildId, ctx);
 
                 ChangeState(BuildId, TIndexBuildInfo::EState::Applying);
                 Progress(BuildId);
@@ -507,8 +511,8 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
     }
 
     void DoComplete(const TActorContext& ctx) override {
-        for (auto& x: ToTabletSend) {
-            Self->IndexBuildPipes.Create(BuildId, std::get<0>(x), std::move(std::get<2>(x)), ctx);
+        for (auto& [shardId, ev]: ToTabletSend) {
+            Self->IndexBuildPipes.Send(BuildId, shardId, std::move(ev), ctx);
         }
         ToTabletSend.clear();
     }
@@ -620,7 +624,7 @@ struct TSchemeShard::TIndexBuilder::TTxReply: public TSchemeShard::TIndexBuilder
         const auto& tabletId = PipeRetry.TabletId;
         const auto& shardIdx = Self->GetShardIdx(tabletId);
 
-        LOG_I("TTxReply : PipeRetry"
+        LOG_N("TTxReply : PipeRetry"
               << ", buildIndexId# " << buildId
               << ", tabletId# " << tabletId
               << ", shardIdx# " << shardIdx);
@@ -999,8 +1003,8 @@ struct TSchemeShard::TIndexBuilder::TTxReply: public TSchemeShard::TIndexBuilder
               << ", BuildIndexId: " << buildInfo->Id
               << ", status: " << Ydb::StatusIds::StatusCode_Name(status)
               << ", error: " << buildInfo->Issue
-              << ", replyTo: " << buildInfo->CreateSender.ToString());
-        LOG_D("Message:\n" << responseEv->Record.ShortDebugString());
+              << ", replyTo: " << buildInfo->CreateSender.ToString()
+              << ", message: " << responseEv->Record.ShortDebugString());
 
         Send(buildInfo->CreateSender, std::move(responseEv), 0, buildInfo->SenderCookie);
     }

ydb/core/tx/schemeshard/schemeshard_cdc_stream_scan.cpp

@@ -114,7 +114,7 @@ struct TSchemeShard::TCdcStreamScan::TTxProgress: public TTransactionBase<TSchem
 
     void Complete(const TActorContext& ctx) override {
         for (auto& [streamPathId, tabletId, ev] : ScanRequests) {
-            Self->CdcStreamScanPipes.Create(streamPathId, tabletId, std::move(ev), ctx);
+            Self->CdcStreamScanPipes.Send(streamPathId, tabletId, std::move(ev), ctx);
         }
 
         if (StreamToProgress) {