Remove KIKIMR_ALLOW_FLOWCONTROLLED_QUEUE_FOR_SSLOOKUP (#6511)

Author: maximyurchuk

ydb/core/base/compile_time_flags.h

@@ -8,11 +8,6 @@
 #define KIKIMR_VDISK_SYNCLOG_ENTRY_POINT_PROTO_FORMAT 0
 #endif
 
-// This feature flag enables use of flow controlled queue in statestorage lookup requests
-#ifndef KIKIMR_ALLOW_FLOWCONTROLLED_QUEUE_FOR_SSLOOKUP
-#define KIKIMR_ALLOW_FLOWCONTROLLED_QUEUE_FOR_SSLOOKUP 0
-#endif
-
 // This feature flag enables statestorage replica probes
 #ifndef KIKIMR_ALLOW_SSREPLICA_PROBES
 #define KIKIMR_ALLOW_SSREPLICA_PROBES 0

ydb/core/base/statestorage_proxy.cpp

@@ -8,7 +8,6 @@
 #include <ydb/library/actors/core/hfunc.h>
 #include <ydb/library/actors/core/interconnect.h>
 #include <ydb/library/actors/core/log.h>
-#include <ydb/library/actors/helpers/flow_controlled_queue.h>
 
 #include <util/digest/city.h>
 #include <util/generic/xrange.h>
@@ -235,15 +234,11 @@ class TStateStorageProxyRequest : public TActor<TStateStorageProxyRequest> {
     }
 
     template<typename TEv>
-    void PrepareInit(TEv *ev, bool allowFlowControlled) {
+    void PrepareInit(TEv *ev) {
         TabletID = ev->TabletID;
         Cookie = ev->Cookie;
         ProxyOptions = ev->ProxyOptions;
-
-        if (allowFlowControlled && FlowControlledInfo.Get() && KIKIMR_ALLOW_FLOWCONTROLLED_QUEUE_FOR_SSLOOKUP)
-            SelectRequestReplicas(FlowControlledInfo.Get());
-        else
-            SelectRequestReplicas(Info.Get());
+        SelectRequestReplicas(Info.Get());
     }
 
     // request setup
@@ -253,7 +248,7 @@ class TStateStorageProxyRequest : public TActor<TStateStorageProxyRequest> {
         BLOG_D("ProxyRequest::HandleInit ev: " << msg->ToString());
         Source = ev->Sender;
 
-        PrepareInit(msg, true);
+        PrepareInit(msg);
         SendRequest([this](ui64 cookie) { return new TEvStateStorage::TEvReplicaLookup(TabletID, cookie); });
 
         Become(&TThis::StateLookup, TDuration::MicroSeconds(StateStorageRequestTimeout), new TEvents::TEvWakeup());
@@ -264,7 +259,7 @@ class TStateStorageProxyRequest : public TActor<TStateStorageProxyRequest> {
         BLOG_D("ProxyRequest::HandleInit ev: %s" << msg->ToString());
         Source = ev->Sender;
 
-        PrepareInit(msg, false);
+        PrepareInit(msg);
 
         SuggestedLeader = msg->ProposedLeader;
         SuggestedLeaderTablet = msg->ProposedLeaderTablet;
@@ -286,7 +281,7 @@ class TStateStorageProxyRequest : public TActor<TStateStorageProxyRequest> {
         BLOG_D("ProxyRequest::HandleInit ev: " << msg->ToString());
         Source = ev->Sender;
 
-        PrepareInit(msg, false);
+        PrepareInit(msg);
 
         SuggestedLeader = msg->ProposedLeader;
         SuggestedGeneration = msg->ProposedGeneration;
@@ -866,13 +861,11 @@ class TStateStorageProxy : public TActor<TStateStorageProxy> {
 
     void Handle(TEvStateStorage::TEvUpdateGroupConfig::TPtr &ev) {
         auto *msg = ev->Get();
-        TIntrusivePtr<TStateStorageInfo> old = Info;
-
         Info = msg->GroupConfig;
         BoardInfo = msg->BoardConfig;
         SchemeBoardInfo = msg->SchemeBoardConfig;
 
-        RegisterDerivedServices(TlsActivationContext->ExecutorThread.ActorSystem, old.Get());
+        RegisterReplicaProbes(TlsActivationContext->ExecutorThread.ActorSystem);
 
         for (const auto& [key, value] : Subscriptions) {
             const auto& [sender, cookie] = key;
@@ -885,11 +878,6 @@ class TStateStorageProxy : public TActor<TStateStorageProxy> {
         }
     }
 
-    void RegisterDerivedServices(TActorSystem *sys, const TStateStorageInfo *old) {
-        RegisterReplicaProbes(sys);
-        RegisterFlowContolled(sys, old);
-    }
-
     void RegisterReplicaProbes(TActorSystem *sys) {
         if (!KIKIMR_ALLOW_SSREPLICA_PROBES)
             return;
@@ -916,65 +904,8 @@ class TStateStorageProxy : public TActor<TStateStorageProxy> {
         Send(ev->Sender, reply.Release(), 0, ev->Cookie);
     }
 
-    void RegisterFlowContolled(TActorSystem *sys, const TStateStorageInfo *old) {
-        if (!KIKIMR_ALLOW_FLOWCONTROLLED_QUEUE_FOR_SSLOOKUP)
-            return;
-
-        TIntrusivePtr<TStateStorageInfo> updated = new TStateStorageInfo();
-        updated->NToSelect = Info->NToSelect;
-        updated->Rings.resize(Info->Rings.size());
-
-        const bool checkOldInfo = FlowControlledInfo && old
-            && updated->NToSelect == FlowControlledInfo->NToSelect
-            && updated->Rings.size() == FlowControlledInfo->Rings.size();
-
-        ui32 ringIdx = 0;
-        for (const ui32 ringsSz = Info->Rings.size(); ringIdx < ringsSz; ++ringIdx) {
-            const bool checkRing = checkOldInfo && (FlowControlledInfo->Rings[ringIdx].Replicas.size() == Info->Rings[ringIdx].Replicas.size());
-
-            TStateStorageInfo::TRing &ctring = updated->Rings[ringIdx];
-            TStateStorageInfo::TRing *fcring = checkRing ? &FlowControlledInfo->Rings[ringIdx] : nullptr;
-            const auto &srcring = Info->Rings[ringIdx];
-            const auto *oldring = checkRing ? &old->Rings[ringIdx] : nullptr;
-
-            ctring.Replicas.resize(srcring.Replicas.size());
-            ui32 replicaIdx = 0;
-            for (const ui32 srcSize = srcring.Replicas.size(); replicaIdx < srcSize; ++replicaIdx) {
-                if (checkRing && srcring.Replicas[replicaIdx] == oldring->Replicas[replicaIdx]) {
-                    ctring.Replicas[replicaIdx] = fcring->Replicas[replicaIdx];
-                    fcring->Replicas[replicaIdx] = TActorId();
-                } else {
-                    if (fcring && replicaIdx < fcring->Replicas.size())
-                        Send(fcring->Replicas[replicaIdx], new TEvents::TEvPoison());
-
-                    TFlowControlledQueueConfig flowConfig;
-                    flowConfig.MaxAllowedInFly = 10000;
-                    flowConfig.TargetDynamicRate = 250000;
-
-                    ctring.Replicas[replicaIdx] = sys->Register(
-                        CreateFlowControlledRequestQueue(srcring.Replicas[replicaIdx], NKikimrServices::TActivity::SS_PROXY_REQUEST, flowConfig),
-                        TMailboxType::ReadAsFilled
-                    );
-                }
-            }
-            if (fcring) {
-                for (const ui32 fcSize = fcring->Replicas.size(); replicaIdx < fcSize; ++replicaIdx) {
-                    Send(fcring->Replicas[replicaIdx], new TEvents::TEvPoison());
-                }
-            }
-        }
-        if (FlowControlledInfo) {
-            for (const ui32 oldSize = FlowControlledInfo->Rings.size(); ringIdx < oldSize; ++ringIdx) {
-                for (TActorId outdated : FlowControlledInfo->Rings[oldSize].Replicas)
-                    Send(outdated, new TEvents::TEvPoison());
-            }
-        }
-
-        FlowControlledInfo = std::move(updated);
-    }
-
     void Registered(TActorSystem* sys, const TActorId&) {
-        RegisterDerivedServices(sys, nullptr);
+        RegisterReplicaProbes(sys);
     }
 public:
     static constexpr NKikimrServices::TActivity::EType ActorActivityType() {