Merge d3aee13 into 6123ecb

Author: gridnevvvit

ydb/core/kqp/counters/kqp_counters.cpp

@@ -776,7 +776,10 @@ TKqpCounters::TKqpCounters(const ::NMonitoring::TDynamicCounterPtr& counters, co
     RmExternalMemory = KqpGroup->GetCounter("RM/ExternalMemory", false);
     RmNotEnoughMemory = KqpGroup->GetCounter("RM/NotEnoughMemory", true);
     RmNotEnoughComputeActors = KqpGroup->GetCounter("RM/NotEnoughComputeActors", true);
+    RmOnStartAllocs = KqpGroup->GetCounter("Rm/OnStartAllocs", true);
     RmExtraMemAllocs = KqpGroup->GetCounter("RM/ExtraMemAllocs", true);
+    RmExtraMemFree = KqpGroup->GetCounter("RM/ExtraMemFree", true);
+    RmOnCompleteFree = KqpGroup->GetCounter("RM/OnCompleteFree", true);
     RmInternalError = KqpGroup->GetCounter("RM/InternalError", true);
     RmSnapshotLatency = KqpGroup->GetHistogram(
         "RM/SnapshotLatency", NMonitoring::ExponentialHistogram(20, 2, 1));

ydb/core/kqp/counters/kqp_counters.h

@@ -350,7 +350,7 @@ class TKqpCounters : public TKqpCountersBase, public NYql::NDq::TSpillingCounter
     ::NMonitoring::TDynamicCounterPtr WorkloadManagerGroup;
 
     ::NMonitoring::TDynamicCounters::TCounterPtr FullScansExecuted;
-    
+
     // Lease updates counters
     ::NMonitoring::THistogramPtr LeaseUpdateLatency;
     ::NMonitoring::THistogramPtr RunActorLeaseUpdateBacklog;
@@ -377,6 +377,9 @@ class TKqpCounters : public TKqpCountersBase, public NYql::NDq::TSpillingCounter
     ::NMonitoring::TDynamicCounters::TCounterPtr RmNotEnoughMemory;
     ::NMonitoring::TDynamicCounters::TCounterPtr RmNotEnoughComputeActors;
     ::NMonitoring::TDynamicCounters::TCounterPtr RmExtraMemAllocs;
+    ::NMonitoring::TDynamicCounters::TCounterPtr RmOnStartAllocs;
+    ::NMonitoring::TDynamicCounters::TCounterPtr RmExtraMemFree;
+    ::NMonitoring::TDynamicCounters::TCounterPtr RmOnCompleteFree;
     ::NMonitoring::TDynamicCounters::TCounterPtr RmInternalError;
     NMonitoring::THistogramPtr RmSnapshotLatency;
     NMonitoring::THistogramPtr NodeServiceStartEventDelivery;

ydb/core/kqp/rm_service/kqp_rm_service.cpp

@@ -264,6 +264,10 @@ class TKqpResourceManager : public IKqpResourceManager {
         // all other requests are not guaranteed to be satisfied.
         // In the nearest future we need to implement several layers of memory requests.
         bool isFirstAllocationRequest = (resources.ExecutionUnits > 0 && resources.MemoryPool == EKqpMemoryPool::DataQuery);
+        if (resources.ExecutionUnits > 0) {
+            Counters->RmOnStartAllocs->Inc();
+        }
+
         if (isFirstAllocationRequest) {
             auto& txBucket = TxBucket(txId);
             with_lock(txBucket.Lock) {
@@ -370,6 +374,12 @@ class TKqpResourceManager : public IKqpResourceManager {
 
         auto& txBucket = TxBucket(txId);
 
+        if (resources.ReleaseAllResources) {
+            Counters->RmOnCompleteFree->Inc();
+        } else {
+            Counters->RmExtraMemFree->Inc();
+        }
+
         {
             TMaybe<TGuard<TMutex>> guard;
             guard.ConstructInPlace(txBucket.Lock);
@@ -385,6 +395,7 @@ class TKqpResourceManager : public IKqpResourceManager {
                 return;
             }
 
+            guard.Clear();
             auto& task = taskIt->second;
             if (resources.ReleaseAllResources && task.ExecutionUnits) {
                 FreeExecutionUnits(task.ExecutionUnits);
@@ -400,10 +411,7 @@ class TKqpResourceManager : public IKqpResourceManager {
             }
 
             task.ScanQueryMemory -= releaseScanQueryMemory;
-            tx.TxScanQueryMemory -= releaseScanQueryMemory;
-
             task.ExternalDataQueryMemory -= releaseExternalDataQueryMemory;
-            tx.TxExternalDataQueryMemory -= releaseExternalDataQueryMemory;
 
             if (task.ScanQueryMemory == 0) {
                 if (task.ResourceBrokerTaskId) {
@@ -419,6 +427,9 @@ class TKqpResourceManager : public IKqpResourceManager {
                 Y_DEBUG_ABORT_UNLESS(reduced);
             }
 
+            guard.ConstructInPlace(txBucket.Lock);
+            tx.TxScanQueryMemory -= releaseScanQueryMemory;
+            tx.TxExternalDataQueryMemory -= releaseExternalDataQueryMemory;
             if (resources.ExecutionUnits) {
                 ui64 remainsTasks = tx.Tasks.size() - 1;
                 if (remainsTasks == 0) {
@@ -428,16 +439,20 @@ class TKqpResourceManager : public IKqpResourceManager {
                 }
             }
 
+            guard.Clear();
+
             i64 prev = ExternalDataQueryMemory.fetch_sub(releaseExternalDataQueryMemory);
             Counters->RmExternalMemory->Sub(releaseExternalDataQueryMemory);
             Y_DEBUG_ABORT_UNLESS(prev >= 0);
             Counters->RmMemory->Sub(releaseScanQueryMemory);
             Y_DEBUG_ABORT_UNLESS(Counters->RmMemory->Val() >= 0);
         } // with_lock (txBucket.Lock)
 
-        with_lock (Lock) {
-            ScanQueryMemoryResource.Release(releaseScanQueryMemory);
-        } // with_lock (Lock)
+        if (releaseScanQueryMemory > 0) {
+            with_lock (Lock) {
+                ScanQueryMemoryResource.Release(releaseScanQueryMemory);
+            } // with_lock (Lock)
+        }
 
         LOG_AS_D("TxId: " << txId << ", taskId: " << taskId << ". Released resources, "
             << "ScanQueryMemory: " << releaseScanQueryMemory << ", "