YQ-3458 fix requests starts for queue size zero

ydb/core/kqp/workload_service/actors/pool_handlers_acors.cpp

@@ -577,7 +577,7 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
     }
 
     void OnScheduleRequest(TRequest* request) override {
-        if (PendingRequests.size() >= MAX_PENDING_REQUESTS || GetLocalSessionsCount() - GetLocalInFlight() > QueueSizeLimit + 1) {
+        if (PendingRequests.size() >= MAX_PENDING_REQUESTS || SaturationSub(GetLocalSessionsCount() - GetLocalInFlight(), InFlightLimit) > QueueSizeLimit) {
             ReplyContinue(request, Ydb::StatusIds::OVERLOADED, TStringBuilder() << "Too many pending requests for pool " << PoolId);
             return;
         }
@@ -695,8 +695,8 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
         size_t delayedRequestsCount = DelayedRequests.size();
         DoStartPendingRequest(GetLoadCpuThreshold());
 
-        if (GlobalState.DelayedRequests + PendingRequests.size() > QueueSizeLimit) {
-            RemoveBackRequests(PendingRequests, std::min(GlobalState.DelayedRequests + PendingRequests.size() - QueueSizeLimit, PendingRequests.size()), [this](TRequest* request) {
+        if (const ui64 delayedRequests = SaturationSub(GlobalState.AmountRequests() + PendingRequests.size(), InFlightLimit); delayedRequests > QueueSizeLimit) {
+            RemoveBackRequests(PendingRequests, std::min(delayedRequests - QueueSizeLimit, PendingRequests.size()), [this](TRequest* request) {
                 ReplyContinue(request, Ydb::StatusIds::OVERLOADED, TStringBuilder() << "Too many pending requests for pool " << PoolId);
             });
             FifoCounters.PendingRequestsCount->Set(PendingRequests.size());
@@ -847,7 +847,7 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
         }
 
         bool canStartRequest = QueueSizeLimit == 0 && GlobalState.RunningRequests < InFlightLimit;
-        canStartRequest |= !GetLoadCpuThreshold() && NodeCount && GlobalState.RunningRequests + NodeCount < InFlightLimit;
+        canStartRequest |= !GetLoadCpuThreshold() && DelayedRequests.size() + GlobalState.DelayedRequests == 0 && NodeCount && GlobalState.RunningRequests + NodeCount < InFlightLimit;
         if (!PendingRequests.empty() && canStartRequest) {
             RunningOperation = true;
             const TString& sessionId = PopPendingRequest();

ydb/core/kqp/workload_service/common/helpers.cpp

@@ -20,4 +20,8 @@ void ParsePoolSettings(const NKikimrSchemeOp::TResourcePoolDescription& descript
     }
 }
 
+ui64 SaturationSub(ui64 x, ui64 y) {
+    return (x > y) ? x - y : 0;
+}
+
 }  // NKikimr::NKqp::NWorkload

ydb/core/kqp/workload_service/common/helpers.h

@@ -99,4 +99,6 @@ NYql::TIssues GroupIssues(const NYql::TIssues& issues, const TString& message);
 
 void ParsePoolSettings(const NKikimrSchemeOp::TResourcePoolDescription& description, NResourcePool::TPoolSettings& poolConfig);
 
+ui64 SaturationSub(ui64 x, ui64 y);
+
 }  // NKikimr::NKqp::NWorkload

ydb/core/kqp/workload_service/ut/common/kqp_workload_service_ut_common.cpp

@@ -453,9 +453,10 @@ class TWorkloadServiceYdbSetup : public IYdbSetup {
             auto subgroup = GetWorkloadManagerCounters(nodeIndex)
                 ->GetSubgroup("pool", CanonizePath(TStringBuilder() << Settings_.DomainName_ << "/" << (poolId ? poolId : Settings_.PoolId_)));
 
-            CheckCommonCounters(subgroup);
+            const TString description = TStringBuilder() << "Node index: " << nodeIndex;
+            CheckCommonCounters(subgroup, description);
             if (checkTableCounters) {
-                CheckTableCounters(subgroup);
+                CheckTableCounters(subgroup, description);
             }
         }
     }
@@ -497,21 +498,21 @@ class TWorkloadServiceYdbSetup : public IYdbSetup {
             ->GetSubgroup("subsystem", "workload_manager");
     }
 
-    static void CheckCommonCounters(NMonitoring::TDynamicCounterPtr subgroup) {
-        UNIT_ASSERT_VALUES_EQUAL(subgroup->GetCounter("LocalInFly", false)->Val(), 0);
-        UNIT_ASSERT_VALUES_EQUAL(subgroup->GetCounter("LocalDelayedRequests", false)->Val(), 0);
-        UNIT_ASSERT_VALUES_EQUAL(subgroup->GetCounter("ContinueOverloaded", true)->Val(), 0);
-        UNIT_ASSERT_VALUES_EQUAL(subgroup->GetCounter("ContinueError", true)->Val(), 0);
-        UNIT_ASSERT_VALUES_EQUAL(subgroup->GetCounter("CleanupError", true)->Val(), 0);
-        UNIT_ASSERT_VALUES_EQUAL(subgroup->GetCounter("Cancelled", true)->Val(), 0);
+    static void CheckCommonCounters(NMonitoring::TDynamicCounterPtr subgroup, const TString& description) {
+        UNIT_ASSERT_VALUES_EQUAL_C(subgroup->GetCounter("LocalInFly", false)->Val(), 0, description);
+        UNIT_ASSERT_VALUES_EQUAL_C(subgroup->GetCounter("LocalDelayedRequests", false)->Val(), 0, description);
+        UNIT_ASSERT_VALUES_EQUAL_C(subgroup->GetCounter("ContinueOverloaded", true)->Val(), 0, description);
+        UNIT_ASSERT_VALUES_EQUAL_C(subgroup->GetCounter("ContinueError", true)->Val(), 0, description);
+        UNIT_ASSERT_VALUES_EQUAL_C(subgroup->GetCounter("CleanupError", true)->Val(), 0, description);
+        UNIT_ASSERT_VALUES_EQUAL_C(subgroup->GetCounter("Cancelled", true)->Val(), 0, description);
 
-        UNIT_ASSERT_GE(subgroup->GetCounter("ContinueOk", true)->Val(), 1);
-        UNIT_ASSERT_VALUES_EQUAL(subgroup->GetCounter("ContinueOk", true)->Val(), subgroup->GetCounter("CleanupOk", true)->Val());
+        UNIT_ASSERT_GE_C(subgroup->GetCounter("ContinueOk", true)->Val(), 1, description);
+        UNIT_ASSERT_VALUES_EQUAL_C(subgroup->GetCounter("ContinueOk", true)->Val(), subgroup->GetCounter("CleanupOk", true)->Val(), description);
     }
 
-    static void CheckTableCounters(NMonitoring::TDynamicCounterPtr subgroup) {
-        UNIT_ASSERT_VALUES_EQUAL(subgroup->GetCounter("PendingRequestsCount", false)->Val(), 0);
-        UNIT_ASSERT_VALUES_EQUAL(subgroup->GetCounter("FinishingRequestsCount", false)->Val(), 0);
+    static void CheckTableCounters(NMonitoring::TDynamicCounterPtr subgroup, const TString& description) {
+        UNIT_ASSERT_VALUES_EQUAL_C(subgroup->GetCounter("PendingRequestsCount", false)->Val(), 0, description);
+        UNIT_ASSERT_VALUES_EQUAL_C(subgroup->GetCounter("FinishingRequestsCount", false)->Val(), 0, description);
 
         const std::vector<std::pair<TString, bool>> tableQueries = {
             {"TCleanupTablesQuery", false},
@@ -524,9 +525,9 @@ class TWorkloadServiceYdbSetup : public IYdbSetup {
         for (const auto& [operation, runExpected] : tableQueries) {
             auto operationSubgroup = subgroup->GetSubgroup("operation", operation);
 
-            UNIT_ASSERT_VALUES_EQUAL_C(operationSubgroup->GetCounter("FinishError", true)->Val(), 0, TStringBuilder() << "Unexpected vaule for operation " << operation);
+            UNIT_ASSERT_VALUES_EQUAL_C(operationSubgroup->GetCounter("FinishError", true)->Val(), 0, TStringBuilder() << description << ", unexpected vaule for operation " << operation);
             if (runExpected) {
-                UNIT_ASSERT_GE_C(operationSubgroup->GetCounter("FinishOk", true)->Val(), 1, TStringBuilder() << "Unexpected vaule for operation " << operation);
+                UNIT_ASSERT_GE_C(operationSubgroup->GetCounter("FinishOk", true)->Val(), 1, TStringBuilder() << description << ", unexpected vaule for operation " << operation);
             }
         }
     }

ydb/core/kqp/workload_service/ut/kqp_workload_service_ut.cpp

@@ -123,6 +123,36 @@ Y_UNIT_TEST_SUITE(KqpWorkloadService) {
         TSampleQueries::TSelect42::CheckResult(hangingRequest.GetResult());
     }
 
+    Y_UNIT_TEST(TestZeroQueueSizeManyQueries) {
+        const i32 inFlight = 10;
+        auto ydb = TYdbSetupSettings()
+            .ConcurrentQueryLimit(inFlight)
+            .QueueSize(0)
+            .QueryCancelAfter(FUTURE_WAIT_TIMEOUT * inFlight)
+            .Create();
+
+        auto settings = TQueryRunnerSettings().HangUpDuringExecution(true);
+
+        std::vector<TQueryRunnerResultAsync> asyncResults;
+        for (size_t i = 0; i < inFlight; ++i) {
+            asyncResults.emplace_back(ydb->ExecuteQueryAsync(TSampleQueries::TSelect42::Query, settings));
+        }
+
+        for (const auto& asyncResult : asyncResults) {
+            ydb->WaitQueryExecution(asyncResult);
+        }
+
+        TSampleQueries::CheckOverloaded(
+            ydb->ExecuteQuery(TSampleQueries::TSelect42::Query, TQueryRunnerSettings().ExecutionExpected(false)),
+            ydb->GetSettings().PoolId_
+        );
+
+        for (const auto& asyncResult : asyncResults) {
+            ydb->ContinueQueryExecution(asyncResult);
+            TSampleQueries::TSelect42::CheckResult(asyncResult.GetResult());
+        }
+    }
+
     Y_UNIT_TEST(TestQueryCancelAfterUnlimitedPool) {
         auto ydb = TYdbSetupSettings()
             .QueryCancelAfter(TDuration::Seconds(10))