Taks #10118: Added timers for TABTranspose, and use 1 thread for data transpose in outputproc

RTCP/Cobalt/CoInterface/src/BudgetTimer.cc

@@ -41,7 +41,7 @@ namespace LOFAR {
 
     BudgetTimer::~BudgetTimer()
     {
-      if (print_on_destruction) {
+      if (print_on_destruction && budget > 0.0) {
         const double realTimePerc = 100.0 * getAverage() / budget;
 
         if (log_on_destruction) {

RTCP/Cobalt/CoInterface/src/TABTranspose.cc

@@ -30,6 +30,7 @@
 #include <Common/Timer.h>
 #include <ApplCommon/PVSSDatapointDefs.h>
 #include <CoInterface/TimeFuncs.h>
+#include <CoInterface/BudgetTimer.h>
 
 using namespace std;
 using boost::format;
@@ -156,7 +157,14 @@ void Block::write( BeamformedData &output ) {
    * The implemented strategy coalesces the writes in batches. Large batches
    * will cause TLB misses due to the larger number of arrays we need to index.
    */
-  const size_t MAXBATCHSIZE = 4;
+
+  // On CEP2, optimal is MAXBATCHSIZE=4 and using 4 threads (in the for loop)
+  // On CEP4, optimal is MAXBATCHSIZE=488 and using 1 thread.
+  //
+  // We want real-time performance, (see tTABTranspose | grep write, should produce
+  // numbers <1s), but if we do not need multiple threads, doing the work in one
+  // is more efficient.
+  const size_t MAXBATCHSIZE = 488;
 
   // Stride between samples in output
   const ptrdiff_t dst_sample_stride = output.samples.strides()[0];
@@ -165,8 +173,6 @@ void Block::write( BeamformedData &output ) {
 
   const vector<float> zeroes(nrChannels * nrSamples, 0.0f);
 
-  // Four threads gives the best performance on CEP2, see figures given by 'tTABTranspose | grep write'
-# pragma omp parallel for num_threads(4)
   for (size_t subbandBase = 0; subbandBase < subbandCache.size(); subbandBase += MAXBATCHSIZE) {
     // Determine actual batch size
     const size_t BATCHSIZE = std::min(MAXBATCHSIZE, subbandCache.size() - subbandBase);
@@ -235,7 +241,7 @@ bool Block::complete() const {
 
 // The BlockCollector collects blocks from different rtcp processes for a TAB.
 // More precisely, we have one BlockCollector per file (i.e. part).
-BlockCollector::BlockCollector( Pool<BeamformedData> &outputPool, size_t fileIdx, size_t nrSubbands, size_t nrChannels, size_t nrSamples, size_t nrBlocks, size_t maxBlocksInFlight, const std::string &logPrefix )
+BlockCollector::BlockCollector( Pool<BeamformedData> &outputPool, size_t fileIdx, size_t nrSubbands, size_t nrChannels, size_t nrSamples, size_t nrBlocks, size_t maxBlocksInFlight, const std::string &logPrefix, double blockDuration )
 :
   logPrefix(logPrefix + "[BlockCollector] "),
 
@@ -256,6 +262,8 @@ BlockCollector::BlockCollector( Pool<BeamformedData> &outputPool, size_t fileIdx
   nrChannels(nrChannels),
   nrSamples(nrSamples),
 
+  blockDuration(blockDuration),
+
   maxBlocksInFlight(maxBlocksInFlight),
   canDrop(maxBlocksInFlight > 0),
   lastEmitted(-1),
@@ -285,7 +293,14 @@ void BlockCollector::addSubband( SmartPtr<Subband> &subband ) {
 void BlockCollector::inputLoop() {
   SmartPtr<Subband> subband;
 
+  BudgetTimer processTimer(
+    "BlockCollector::inputLoop: weaving subbands",
+    blockDuration / nrSubbands,
+    true, true);
+
   while ((subband = inputQueue.remove()) != NULL) {
+    BudgetTimer::StartStop ss(processTimer);
+
     processSubband(subband);
   }
 }
@@ -294,14 +309,18 @@ void BlockCollector::inputLoop() {
 void BlockCollector::outputLoop() {
   SmartPtr<Block> block;
 
-  NSTimer writeTimer("Block: data transpose/zeroing", true, true);
+  BudgetTimer writeTimer(
+    "BlockCollector::outputLoop: data transpose/zeroing",
+    blockDuration,
+    true, true);
 
   while ((block = outputQueue.remove()) != NULL) {
     SmartPtr<BeamformedData> output = outputPool.free.remove();
 
-    writeTimer.start();
-    block->write(*output);
-    writeTimer.stop();
+    {
+      BudgetTimer::StartStop ss(writeTimer);
+      block->write(*output);
+    }
 
     outputPool.filled.append(output);
   }

RTCP/Cobalt/CoInterface/src/TABTranspose.h

@@ -182,7 +182,7 @@ namespace LOFAR
          * maxBlocksInFlight: the maximum number of blocks to process in
          *                    parallel (or 0 for no limit).
          */
-        BlockCollector( Pool<BeamformedData> &outputPool, size_t fileIdx, size_t nrSubbands, size_t nrChannels, size_t nrSamples, size_t nrBlocks = 0, size_t maxBlocksInFlight = 0, const std::string &logPrefix = "" );
+        BlockCollector( Pool<BeamformedData> &outputPool, size_t fileIdx, size_t nrSubbands, size_t nrChannels, size_t nrSamples, size_t nrBlocks = 0, size_t maxBlocksInFlight = 0, const std::string &logPrefix = "", double blockDuration = 0.0 );
 
         ~BlockCollector();
 
@@ -227,6 +227,8 @@ namespace LOFAR
         const size_t nrChannels;
         const size_t nrSamples;
 
+        const double blockDuration;
+
         // upper limit for blocks.size(), or 0 if unlimited
         const size_t maxBlocksInFlight;
 

RTCP/Cobalt/OutputProc/src/GPUProcIO.cc

@@ -233,7 +233,7 @@ bool process(Stream &controlStream)
         // #blocks here is the number of blocks that can be collected in parallel on
         // the input side (what the correlator sends)
         collectors[fileIdx] = new TABTranspose::BlockCollector(
-          *outputPools[fileIdx], fileIdx, nrSubbands, nrChannels, nrSamples, parset.settings.nrBlocks(), parset.settings.realTime ? 5 : 0, logPrefix);
+          *outputPools[fileIdx], fileIdx, nrSubbands, nrChannels, nrSamples, parset.settings.nrBlocks(), parset.settings.realTime ? 5 : 0, logPrefix, parset.settings.blockDuration());
 
         LOG_DEBUG_STR("Setting up writer for " << file.location.filename);