HPCC-32931 Capture and report execute timings for splitters

Signed-off-by: Shamser Ahmed <[email protected]>
hpcc-systems · Nov 4, 2024 · f6730cb · f6730cb
1 parent 4c31ec4
commit f6730cb
Showing 1 changed file with 8 additions and 1 deletion.
diff --git a/thorlcr/activities/nsplitter/thnsplitterslave.cpp b/thorlcr/activities/nsplitter/thnsplitterslave.cpp
@@ -101,6 +101,7 @@ class NSplitterSlaveActivity : public CSlaveActivity, implements ISharedSmartBuf
     bool inputConnected = false;
     unsigned numOutputs = 0;
     ThorDataLinkMetaInfo cachedMetaInfo;
+    std::unique_ptr<BlockedActivityTimer> blockedActivityTimer;
 
     // NB: CWriter only used by 'balanced' splitter, which blocks write when too far ahead
     class CWriter : public CSimpleInterface, IThreaded
@@ -217,7 +218,10 @@ class NSplitterSlaveActivity : public CSlaveActivity, implements ISharedSmartBuf
                 assertex(((unsigned)-1) != connectedOutputCount);
                 activeOutputCount = connectedOutputCount;
 
-                PARENT::start();
+                {
+                    ActivityTimer t(slaveTimerStats, queryTimeActivities());
+                    PARENT::start();
+                }
                 initMetaInfo(cachedMetaInfo);
                 cachedMetaInfo.suppressLookAhead = spill; // only suppress downstream lookaheads if this is a spilling splitter
 
@@ -299,6 +303,7 @@ class NSplitterSlaveActivity : public CSlaveActivity, implements ISharedSmartBuf
     }
     inline const void *nextRow(unsigned outIdx, rowcount_t current)
     {
+        ActivityTimer t(slaveTimerStats, queryTimeActivities());
         if (1 == activeOutputCount) // will be true, if only 1 input connected, or only 1 input was active (others stopped) when it started reading
             return inputStream->nextRow();
         if (recsReady == current && writeAheadException.get())
@@ -412,11 +417,13 @@ class NSplitterSlaveActivity : public CSlaveActivity, implements ISharedSmartBuf
     virtual void paged() { pagedOut = true; }
     virtual void blocked()
     {
+        blockedActivityTimer.reset(new BlockedActivityTimer(slaveTimerStats, queryTimeActivities()));
         writeBlocked = true; // Prevent other users getting beyond checking recsReady in writeahead()
         writeAheadCrit.leave();
     }
     virtual void unblocked()
     {
+        blockedActivityTimer.reset(nullptr);
         writeAheadCrit.enter();
         writeBlocked = false;
         if (stalledWriters.ordinality())