Task #11059: Added BudgetTimer around writing output to disk

RTCP/Cobalt/CoInterface/src/BudgetTimer.h

@@ -42,6 +42,18 @@ namespace LOFAR {
 
       void stop();
 
+      // Internal class to do an automatic start/stop.
+      class StartStop {
+      public:
+        StartStop(BudgetTimer& timer) : itsTimer(timer) { itsTimer.start(); }
+        ~StartStop() { itsTimer.stop(); }
+        private:
+          // Forbid copy.
+          StartStop (const StartStop&);
+          StartStop& operator= (StartStop&);
+          BudgetTimer& itsTimer;
+      };
+
     private:
         const double budget;
 

RTCP/Cobalt/OutputProc/src/OutputThread.cc

@@ -36,6 +36,7 @@
 #include <CoInterface/OutputTypes.h>
 #include <CoInterface/Exceptions.h>
 #include <CoInterface/LTAFeedback.h>
+#include <CoInterface/BudgetTimer.h>
 
 #if defined HAVE_AIPSPP
 #include <casa/Exceptions/Error.h>
@@ -71,6 +72,7 @@ namespace LOFAR
       itsBlocksDropped(0),
       itsNrExpectedBlocks(0),
       itsNextSequenceNumber(0),
+      itsBlockDuration(parset.settings.blockDuration()),
       itsOutputPool(outputPool)
     {
     }
@@ -97,7 +99,7 @@ namespace LOFAR
         LOG_WARN_STR(itsLogPrefix << "Did not receive " << droppedBlocks << " blocks");
 
         itsMdLogger.log(itsMdKeyPrefix + PN_COP_DROPPED + streamNrStr,
-                        itsBlocksDropped * static_cast<float>(itsParset.settings.blockDuration()));
+                        itsBlocksDropped * static_cast<float>(itsBlockDuration));
       }
 
       itsNextSequenceNumber = data->sequenceNumber() + 1;
@@ -106,15 +108,22 @@ namespace LOFAR
       itsMdLogger.log(itsMdKeyPrefix + PN_COP_DROPPING + streamNrStr,
                       droppedBlocks > 0); // logged too late if dropping: not anymore...
       itsMdLogger.log(itsMdKeyPrefix + PN_COP_WRITTEN  + streamNrStr,
-                      itsBlocksWritten * static_cast<float>(itsParset.settings.blockDuration()));
+                      itsBlocksWritten * static_cast<float>(itsBlockDuration));
     }
 
 
     template<typename T> void OutputThread<T>::doWork()
     {
+      BudgetTimer writeTimer(
+        "writeOutput",
+        itsBlockDuration,
+        true, true);
+
       for (SmartPtr<T> data; (data = itsOutputPool.filled.remove()) != 0; itsOutputPool.free.append(data)) {
         if (itsParset.settings.realTime) {
           try {
+            BudgetTimer::StartStop ss(writeTimer);
+
             itsWriter->write(data);
           } catch (SystemCallException &ex) {
             LOG_WARN_STR(itsLogPrefix << "OutputThread caught non-fatal exception: " << ex.what());
@@ -262,6 +271,7 @@ namespace LOFAR
           logPrefix + "[SubbandOutputThread] ",
           targetDirectory)
     {
+      itsBlockDuration = parset.settings.correlator.integrationTime();
     }
 
 

RTCP/Cobalt/OutputProc/src/OutputThread.h

@@ -97,6 +97,7 @@ namespace LOFAR
       size_t itsBlocksWritten, itsBlocksDropped;
       size_t itsNrExpectedBlocks;
       size_t itsNextSequenceNumber;
+      double itsBlockDuration; // seconds
 
       Pool<T> &itsOutputPool;