Do not call TaskRunner for periodic stats updates (#18143)

Author: Hor911

ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h

@@ -1133,15 +1133,9 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
                 break;
             }
             case EEvWakeupTag::PeriodicStatsTag: {
-                const auto maxInterval = RuntimeSettings.ReportStatsSettings->MaxInterval;
                 if (Running && State == NDqProto::COMPUTE_STATE_EXECUTING) {
-                    if (ProcessOutputsState.LastRunStatus == ERunStatus::Finished) {
-                        // wait until all outputs are drained
-                        ReportStats();
-                    } else {
-                        DoExecute();
-                    }
-                    this->Schedule(maxInterval, new NActors::TEvents::TEvWakeup(EEvWakeupTag::PeriodicStatsTag));
+                    ReportStats();
+                    this->Schedule(RuntimeSettings.ReportStatsSettings->MaxInterval, new NActors::TEvents::TEvWakeup(EEvWakeupTag::PeriodicStatsTag));
                 }
                 break;
             }

ydb/library/yql/dq/actors/compute/dq_compute_actor_stats.cpp

@@ -65,8 +65,14 @@ void FillTaskRunnerStats(ui64 taskId, ui32 stageId, const TDqTaskRunnerStats& ta
 
     protoTask->SetWaitInputTimeUs(taskStats.WaitInputTime.MicroSeconds());
     protoTask->SetWaitOutputTimeUs(taskStats.WaitOutputTime.MicroSeconds());
-    protoTask->SetCurrentWaitInputTimeUs(taskStats.CurrentWaitInputTime.MicroSeconds());
-    protoTask->SetCurrentWaitOutputTimeUs(taskStats.CurrentWaitOutputTime.MicroSeconds());
+
+    auto now = TInstant::Now();
+    if (taskStats.CurrentWaitInputStartTime) {
+        protoTask->SetCurrentWaitInputTimeUs((now - taskStats.CurrentWaitInputStartTime).MicroSeconds());
+    }
+    if (taskStats.CurrentWaitOutputStartTime) {
+        protoTask->SetCurrentWaitOutputTimeUs((now - taskStats.CurrentWaitOutputStartTime).MicroSeconds());
+    }
 
     if (StatsLevelCollectFull(level)) {
         protoTask->SetSpillingComputeWriteBytes(taskStats.SpillingComputeWriteBytes);

ydb/library/yql/dq/runtime/dq_tasks_runner.cpp

@@ -786,27 +786,34 @@ class TDqTaskRunner : public IDqTaskRunner {
         }
 
         if (Y_LIKELY(CollectBasic())) {
+            auto now = TInstant::Now();
             switch (runStatus) {
                 case ERunStatus::Finished:
                     // finished => waiting for nothing
-                    Stats->CurrentWaitInputTime = TDuration::Zero();
-                    Stats->CurrentWaitOutputTime = TDuration::Zero();
-                    Stats->FinishTs = TInstant::Now();
+                    Stats->CurrentWaitInputStartTime = TInstant::Zero();
+                    Stats->CurrentWaitOutputStartTime = TInstant::Zero();
+                    Stats->FinishTs = now;
                     break;
                 case ERunStatus::PendingInput:
                     // output is checked first => not waiting for output
-                    Stats->CurrentWaitOutputTime = TDuration::Zero();
+                    Stats->CurrentWaitOutputStartTime = TInstant::Zero();
                     if (Y_LIKELY(InputConsumed)) {
                         // did smth => waiting for nothing
-                        Stats->CurrentWaitInputTime = TDuration::Zero();
+                        Stats->CurrentWaitInputStartTime = TInstant::Zero();
                     } else {
                         StartWaitingInput();
+                        if (Y_LIKELY(!Stats->CurrentWaitInputStartTime)) {
+                            Stats->CurrentWaitInputStartTime = now;
+                        }
                     }
                     break;
                 case ERunStatus::PendingOutput:
                     // waiting for output => not waiting for input
-                    Stats->CurrentWaitInputTime = TDuration::Zero();
+                    Stats->CurrentWaitInputStartTime = TInstant::Zero();
                     StartWaitingOutput();
+                    if (Y_LIKELY(!Stats->CurrentWaitOutputStartTime)) {
+                        Stats->CurrentWaitOutputStartTime = now;
+                    }
                     break;
             }
         }
@@ -1077,7 +1084,6 @@ class TDqTaskRunner : public IDqTaskRunner {
             } else {
                 Stats->WaitStartTime += delta;
             }
-            Stats->CurrentWaitInputTime += delta;
         }
         StartWaitInputTime = now;
     }
@@ -1087,7 +1093,6 @@ class TDqTaskRunner : public IDqTaskRunner {
         if (Y_LIKELY(StartWaitOutputTime)) {
             auto delta = now - *StartWaitOutputTime;
             Stats->WaitOutputTime += delta;
-            Stats->CurrentWaitOutputTime += delta;
         }
         StartWaitOutputTime = now;
     }
@@ -1101,14 +1106,12 @@ class TDqTaskRunner : public IDqTaskRunner {
             } else {
                 Stats->WaitStartTime += delta;
             }
-            Stats->CurrentWaitInputTime += delta;
             StartWaitInputTime.reset();
             TDuration::Zero();
         }
         if (StartWaitOutputTime) {
             auto delta = now - *StartWaitOutputTime;
             Stats->WaitOutputTime += delta;
-            Stats->CurrentWaitOutputTime += delta;
             StartWaitOutputTime.reset();
         }
     }

ydb/library/yql/dq/runtime/dq_tasks_runner.h

@@ -57,8 +57,9 @@ struct TDqTaskRunnerStats {
     TDuration WaitStartTime;
     TDuration WaitInputTime;
     TDuration WaitOutputTime;
-    TDuration CurrentWaitInputTime;
-    TDuration CurrentWaitOutputTime;
+
+    TInstant CurrentWaitInputStartTime;
+    TInstant CurrentWaitOutputStartTime;
 
     ui64 SpillingComputeWriteBytes;
     ui64 SpillingChannelWriteBytes;