25-1 Allow to send duplicates through TDedicatedPipePool (#18324)

Author: kunga

ydb/core/tx/schemeshard/dedicated_pipe_pool.h

@@ -15,18 +15,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.Register(CreateClient(ctx.SelfID, ui64(dst), TClientRetryPolicy {
-            .MinRetryTime = TDuration::MilliSeconds(100),
-            .MaxRetryTime = TDuration::Seconds(30),
-        }));
+        if (!Pipes[entityId].contains(dst)) {
+            const auto clientId = ctx.Register(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);
     }
 
@@ -53,10 +56,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;
@@ -70,7 +73,7 @@ class TDedicatedPipePool {
             Close(entityId, tabletId, ctx);
         }
 
-        return tablets.size();
+        return;
     }
 
     void Shutdown(const TActorContext& ctx) {

ydb/core/tx/schemeshard/schemeshard_build_index__progress.cpp

@@ -489,7 +489,7 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
 private:
     TIndexBuildId BuildId;
 
-    TDeque<std::tuple<TTabletId, ui64, THolder<IEventBase>>> ToTabletSend;
+    TMap<TTabletId, THolder<IEventBase>> ToTabletSend;
 
     template <typename Record>
     TTabletId CommonFillRecord(Record& record, TShardIdx shardIdx, TIndexBuildInfo& buildInfo) {
@@ -541,9 +541,9 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
 
         auto shardId = CommonFillRecord(ev->Record, shardIdx, buildInfo);
         ev->Record.SetSeed(ui64(shardId));
-        LOG_D("TTxBuildProgress: TEvSampleKRequest: " << ev->Record.ShortDebugString());
+        LOG_N("TTxBuildProgress: TEvSampleKRequest: " << ev->Record.ShortDebugString());
 
-        ToTabletSend.emplace_back(shardId, ui64(BuildId), std::move(ev));
+        ToTabletSend.emplace(shardId, std::move(ev));
     }
 
     void SendKMeansReshuffleRequest(TShardIdx shardIdx, TIndexBuildInfo& buildInfo) {
@@ -585,9 +585,9 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
             r.ClearClusters();
             return r.ShortDebugString();
         };
-        LOG_D("TTxBuildProgress: TEvReshuffleKMeansRequest: " << toDebugStr(ev->Record));
+        LOG_N("TTxBuildProgress: TEvReshuffleKMeansRequest: " << toDebugStr(ev->Record));
 
-        ToTabletSend.emplace_back(shardId, ui64(BuildId), std::move(ev));
+        ToTabletSend.emplace(shardId, std::move(ev));
     }
 
     void SendKMeansLocalRequest(TShardIdx shardIdx, TIndexBuildInfo& buildInfo) {
@@ -637,9 +637,9 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
 
         auto shardId = CommonFillRecord(ev->Record, shardIdx, buildInfo);
         ev->Record.SetSeed(ui64(shardId));
-        LOG_D("TTxBuildProgress: TEvLocalKMeansRequest: " << ev->Record.ShortDebugString());
+        LOG_N("TTxBuildProgress: TEvLocalKMeansRequest: " << ev->Record.ShortDebugString());
 
-        ToTabletSend.emplace_back(shardId, ui64(BuildId), std::move(ev));
+        ToTabletSend.emplace(shardId, std::move(ev));
     }
 
     void SendBuildIndexRequest(TShardIdx shardIdx, TIndexBuildInfo& buildInfo) {
@@ -688,9 +688,9 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
 
         auto shardId = CommonFillRecord(ev->Record, shardIdx, buildInfo);
 
-        LOG_D("TTxBuildProgress: TEvBuildIndexCreateRequest: " << ev->Record.ShortDebugString());
+        LOG_N("TTxBuildProgress: TEvBuildIndexCreateRequest: " << ev->Record.ShortDebugString());
 
-        ToTabletSend.emplace_back(shardId, ui64(BuildId), std::move(ev));
+        ToTabletSend.emplace(shardId, std::move(ev));
     }
 
     void SendUploadSampleKRequest(TIndexBuildInfo& buildInfo) {
@@ -712,6 +712,7 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
         buildInfo.InProgressShards = {};
         buildInfo.ToUploadShards = {};
 
+        ToTabletSend.clear();
         Self->IndexBuildPipes.CloseAll(BuildId, ctx);
     }
 
@@ -968,8 +969,8 @@ struct TSchemeShard::TIndexBuilder::TTxProgress: public TSchemeShard::TIndexBuil
         Y_ABORT_UNLESS(buildInfoPtr);
         auto& buildInfo = *buildInfoPtr->Get();
 
-        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:
@@ -1227,8 +1228,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();
     }
@@ -1303,7 +1304,7 @@ struct TSchemeShard::TIndexBuilder::TTxReplyRetry: public TSchemeShard::TIndexBu
         const auto& tabletId = PipeRetry.TabletId;
         const auto& shardIdx = Self->GetShardIdx(tabletId);
 
-        LOG_I("TTxReply : PipeRetry, id# " << buildId
+        LOG_N("TTxReply : PipeRetry, id# " << buildId
               << ", tabletId# " << tabletId
               << ", shardIdx# " << shardIdx);
 
@@ -2120,8 +2121,8 @@ struct TSchemeShard::TIndexBuilder::TTxReplyModify: public TSchemeShard::TIndexB
               << ", 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

@@ -115,7 +115,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) {