Move MaxTxInFly to ICB (ydb-platform#27635)

Author: neyrox

ydb/core/protos/config.proto

@@ -1629,6 +1629,12 @@ message TImmediateControlsConfig {
             MinValue: 8,
             MaxValue: 4096,
             DefaultValue: 256 }];
+
+        optional uint64 MaxTxInFly = 2 [(ControlOptions) = {
+            Description: "Maximum tx queue size for all tablets",
+            MinValue: 0,
+            MaxValue: 1000000,
+            DefaultValue: 10000 }];
     }
 
     message TDSProxyControls {

ydb/core/protos/memory_controller_config.proto

@@ -28,6 +28,4 @@ message TMemoryControllerConfig {
 
     optional float CompactionLimitPercent = 130 [default = 10];
     optional uint64 CompactionLimitBytes = 131;
-
-    optional uint64 MaxTxInFly = 132 [default = 10000];
 }

ydb/core/tablet_flat/flat_executor.cpp

@@ -133,12 +133,12 @@ TExecutor::TExecutor(
     , Time(TAppData::TimeProvider)
     , Owner(owner)
     , OwnerActorId(ownerActorId)
-    , MaxTxInFly(AppData()->MemoryControllerConfig.GetMaxTxInFly())
     , Emitter(new TIdEmitter)
     , CounterEventsInFlight(new TEvTabletCounters::TInFlightCookie)
     , Stats(new TExecutorStatsImpl())
     , LogFlushDelayOverrideUsec(-1, -1, 60*1000*1000)
     , MaxCommitRedoMB(256, 1, 4096)
+    , MaxTxInFly(10000, 0, 1000000)
 {}
 
 TExecutor::~TExecutor() {
@@ -181,6 +181,7 @@ void TExecutor::Registered(TActorSystem *sys, const TActorId&)
     TString myTabletType = TTabletTypes::TypeToStr(Owner->TabletType());
     AppData()->Icb->RegisterSharedControl(LogFlushDelayOverrideUsec, myTabletType + "_LogFlushDelayOverrideUsec");
     AppData()->Icb->RegisterSharedControl(MaxCommitRedoMB, "TabletControls.MaxCommitRedoMB");
+    AppData()->Icb->RegisterSharedControl(MaxTxInFly, "TabletControls.MaxTxInFly");
 
     // instantiate alert counters so even never reported alerts are created
     GetServiceCounters(AppData()->Counters, "tablets")->GetCounter("alerts_pending_nodata", true);
@@ -4073,8 +4074,7 @@ void TExecutor::ForceSendCounters() {
 
 float TExecutor::GetRejectProbability() const {
     // Limit number of in-flight TXs
-    // TODO: make configurable
-    if (Stats->TxInFly > MaxTxInFly) {
+    if (Stats->TxInFly > ui64(MaxTxInFly)) {
         HadRejectProbabilityByTxInFly = true;
         return 1.0;
     }
@@ -4110,7 +4110,7 @@ float TExecutor::GetRejectProbability() const {
 }
 
 void TExecutor::MaybeRelaxRejectProbability() {
-    if (HadRejectProbabilityByTxInFly && Stats->TxInFly <= MaxTxInFly ||
+    if (HadRejectProbabilityByTxInFly && Stats->TxInFly <= ui64(MaxTxInFly) ||
         HadRejectProbabilityByOverload)
     {
         HadRejectProbabilityByTxInFly = false;

ydb/core/tablet_flat/flat_executor.h

@@ -385,8 +385,6 @@ class TExecutor
     // Counts the number of times LeaseDuration was increased
     size_t LeaseDurationIncreases = 0;
 
-    const ui64 MaxTxInFly;
-
     struct TLeaseCommit : public TIntrusiveListItem<TLeaseCommit> {
         using TByEndMap = std::multimap<TMonotonic, TLeaseCommit*>;
 
@@ -501,6 +499,7 @@ class TExecutor
 
     TControlWrapper LogFlushDelayOverrideUsec;
     TControlWrapper MaxCommitRedoMB;
+    TControlWrapper MaxTxInFly;
 
     TActorId BackupWriter;
 

ydb/tests/olap/test_overloads.py

@@ -5,6 +5,7 @@
 import yatest.common
 import ydb
 import random
+import requests
 
 from ydb.tests.library.harness.kikimr_config import KikimrConfigGenerator
 from ydb.tests.library.harness.kikimr_runner import KiKiMR
@@ -110,13 +111,13 @@ def _setup_ydb_rp(cls):
         logger.info(yatest.common.execute([ydb_path, "-V"], wait=True).stdout.decode("utf-8"))
         config = KikimrConfigGenerator(
             extra_feature_flags={"enable_olap_reject_probability": True},
-            memory_controller_config={"max_tx_in_fly": 0},
         )
         cls.cluster = KiKiMR(config)
         cls.cluster.start()
         node = cls.cluster.nodes[1]
         cls.ydb_client = YdbClient(endpoint=f"grpc://{node.host}:{node.port}", database=f"/{config.domain_name}")
         cls.ydb_client.wait_connection(timeout=60)
+        cls.mon_url = f"http://{node.host}:{node.mon_port}"
 
     def get_row_count(self) -> int:
         return self.ydb_client.query(f"select count(*) as Rows from `{self.table_name}`")[0].rows[0]["Rows"]
@@ -208,8 +209,16 @@ def test_overloads_workload(self, writing_in_flight_requests_count_limit, writin
         logging.info(f"Count rows after insert {self.get_row_count()}")
         assert self.get_row_count() != 0
 
+    def tune_icb(self):
+        response = requests.post(
+            self.mon_url + "/actors/icb",
+            data="TabletControls.MaxTxInFly=0"
+        )
+        response.raise_for_status()
+
     def test_overloads_reject_probability(self):
         self._setup_ydb_rp()
+        self.tune_icb()
 
         table_path = f"{self.ydb_client.database}/table_for_test_overloads_reject_probability"
         self.ydb_client.query(