Task #3566: Write final size and duration of observation to log

LCS/Stream/include/Stream/FileStream.h

@@ -39,6 +39,8 @@ class FileStream : public FileDescriptorBasedStream
     virtual ~FileStream();
 
     virtual void skip( size_t bytes ); // seek ahead
+
+    virtual size_t size(); // return file size
 };
 
 } // namespace LOFAR

LCS/Stream/src/FileStream.cc

@@ -68,4 +68,18 @@ void FileStream::skip(size_t bytes)
     throw SystemCallException("lseek", errno, THROW_ARGS);
 }
 
+size_t FileStream::size()
+{
+  // lseek returning -1 can be either an error, or theoretically,
+  // a valid new file position. To make sure, we need to
+  // clear and check errno.
+  errno = 0;
+  off_t pos = lseek(fd, 0, SEEK_END);
+
+  if (pos == (off_t)-1 && errno)
+    throw SystemCallException("lseek", errno, THROW_ARGS);
+
+  return pos;
+}
+
 } // namespace LOFAR

RTCP/Run/src/RTCP.parset

@@ -33,7 +33,7 @@ OLAP.nrTimesInFrame = 16
 #     [ 3*(0;1;2;3)  ] --> [ 0,1,2,3,0,1,2,3,0,1,2,3 ]
 #     [ 3*(300..303) ] --> [ 300,301,302,303,300,301,302,303,300,301,302,303 ]
 #     [ 2*(5*0)      ] --> [ 0,0,0,0,0,0,0,0,0,0 ]       
-Observation.subbandList  = [100..159]
+Observation.subbandList  = [100..110]
 Observation.sampleClock  = 200
 Observation.nrBitsPerSample = 16
 
@@ -68,8 +68,8 @@ Observation.Beam[0].TiedArrayBeam[0].coherent = T
 OLAP.Storage.hosts = [locus095]
 OLAP.Storage.targetDirectory = /data # will be appended with dirmask and filenammask
 
-Observation.DataProducts.Output_Beamformed.enabled = T
-Observation.DataProducts.Output_Correlated.enabled = F
+Observation.DataProducts.Output_Beamformed.enabled = F
+Observation.DataProducts.Output_Correlated.enabled = T
 Observation.DataProducts.Output_Trigger.enabled = F
 
 Observation.DataProducts.Output_Correlated.namemask = L${OBSID}_SB${SUBBAND}_uv.MS

RTCP/Storage/include/Storage/MSWriter.h

@@ -37,6 +37,8 @@ class MSWriter
     virtual	 ~MSWriter();
 
     virtual void write(StreamableData *) = 0;
+
+    virtual size_t getDataSize();
 };
 
 

RTCP/Storage/include/Storage/MSWriterFile.h

@@ -43,6 +43,8 @@ class MSWriterFile : public MSWriter
 
     virtual void write(StreamableData *data);
 
+    virtual size_t getDataSize();
+
   protected:
     FastFileStream	 itsFile;
 };

RTCP/Storage/src/FastFileStream.cc

@@ -199,6 +199,24 @@ void FastFileStream::skip(size_t bytes)
 }
 
 
+size_t FastFileStream::size()
+{
+  // size we might have skip()ed and have some remaining data to write,
+  // we cannot rely on FileStream::size(), which would report the current
+  // file size, without skips or remainders in our buffer.
+
+  errno = 0;
+
+  off_t curlen = lseek(fd, 0, SEEK_CUR); // NOT SEEK_END, because skip() might push us beyond the end
+
+  // lseek can return -1 as a valid file position, so check errno as well
+  if (curlen == (off_t)-1 && errno)
+    throw SystemCallException("lseek", errno, THROW_ARGS);
+
+  return curlen + remainder;
+}
+
+
 } // namespace RTCP
 } // namespace LOFAR
 

RTCP/Storage/src/MSWriter.cc

@@ -35,6 +35,11 @@ MSWriter::~MSWriter()
 {
 }
 
+size_t MSWriter::getDataSize()
+{
+  return 0;
+}
+
 
 }
 }

RTCP/Storage/src/MSWriterFile.cc

@@ -50,6 +50,12 @@ void MSWriterFile::write(StreamableData *data)
 }
 
 
+size_t MSWriterFile::getDataSize()
+{
+  return itsFile.size();
+}
+
+
 } // namespace RTCP
 } // namespace LOFAR
 

RTCP/Storage/src/OutputThread.cc

@@ -321,6 +321,25 @@ void OutputThread::cleanUp()
   unsigned percent_written = roundedPercentage(itsBlocksWritten, itsNrExpectedBlocks);
 
   LOG_INFO_STR(itsLogPrefix << "Finished writing: " << itsBlocksWritten << " blocks written (" << percent_written << "%), " << itsBlocksDropped << " blocks dropped: " << std::setprecision(3) << dropPercent << "% lost" );
+
+  // log some final characteristics for CEPlogProcessor for feedback to MoM/LTA
+  switch (itsOutputType) {
+    case CORRELATED_DATA:
+      {
+        LOG_INFO_STR(itsLogPrefix << "Final characteristics: "
+            << ", duration "     << setprecision(8) << itsNextSequenceNumber * itsParset.IONintegrationTime() << " s"
+            << ", size "         << itsWriter->getDataSize() << " bytes"
+        );
+      }
+      break;
+    case BEAM_FORMED_DATA:
+      itsNrExpectedBlocks = itsParset.nrBeamFormedBlocks();
+      break;
+
+    default:
+      break;
+  }
+}
 }