Task #11059: Fixed timer for Correlated output for sub-block integrations and when the link drops.

RTCP/Cobalt/GPUProc/src/cuda/Pipelines/Pipeline.cc

@@ -819,13 +819,14 @@ namespace LOFAR
     {
       BudgetTimer writeTimer(
         str(format("[subband %u] write correlated output") % globalSubbandIdx),
-        ps.settings.correlator.integrationTime(),
+        ps.settings.blockDuration(),
         true, true);
 
       // Register our thread to be killable at exit
       OMPThreadSet::ScopedRun sr(outputThreads);
 
       SmartPtr<Stream> outputStream;
+      bool outputEnabled = true;
 
       if (ps.settings.correlator.enabled) {
         const string desc = getStreamDescriptorBetweenIONandStorage(ps, CORRELATED_DATA, globalSubbandIdx,
@@ -835,7 +836,7 @@ namespace LOFAR
           outputStream = createStream(desc, false, 0);
         } catch (Exception &ex) {
           LOG_ERROR_STR("Error writing subband " << globalSubbandIdx << ", dropping all subsequent blocks: " << ex.what());
-          return;
+          outputEnabled = false;
         }
       }
 
@@ -846,23 +847,23 @@ namespace LOFAR
         const struct BlockID id = data->blockID;
         ASSERT( globalSubbandIdx == id.globalSubbandIdx );
 
-        if (data->emit_correlatedData) {
+        if (outputEnabled && data->emit_correlatedData) {
           ASSERT(ps.settings.correlator.enabled);
           ASSERT(outputStream.get());
 
           LOG_DEBUG_STR("[" << id << "] Writing start");
 
           // Write block to outputProc 
+          writeTimer.start();
           try {
-            writeTimer.start();
             for (size_t i = 0; i < data->correlatedData.subblocks.size(); ++i)
               data->correlatedData.subblocks[i]->write(outputStream.get(), true);
-            writeTimer.stop();
           } catch (Exception &ex) {
             // No reconnect, as outputProc doesn't yet re-listen when the conn drops.
             LOG_ERROR_STR("Error writing subband " << id.globalSubbandIdx << ", dropping all subsequent blocks: " << ex.what());
-            return;
+            outputEnabled = false;
           }
+          writeTimer.stop();
 
           if (id.localSubbandIdx == 0 || id.localSubbandIdx == subbandIndices.size() - 1)
             LOG_INFO_STR("[" << id << "] Done");