Task #9052: Create measurementsets in parallel to writing data, for faster startup

RTCP/Cobalt/OutputProc/src/MSWriter.cc

@@ -42,6 +42,10 @@ namespace LOFAR
     {
     }
 
+    void MSWriter::createMetaData()
+    {
+    }
+
     void MSWriter::augment(const FinalMetaData &finalMetaData)
     {
       (void)finalMetaData;

RTCP/Cobalt/OutputProc/src/MSWriter.h

@@ -39,6 +39,8 @@ namespace LOFAR
 
       virtual void write(StreamableData *) = 0;
 
+      virtual void createMetaData();
+
       virtual void augment(const FinalMetaData &finalMetaData);
 
       virtual size_t getDataSize();

RTCP/Cobalt/OutputProc/src/MSWriterCorrelated.cc

@@ -27,7 +27,6 @@
 #include <boost/format.hpp>
 #include <boost/lexical_cast.hpp>
 
-#include <Common/SystemUtil.h>
 #include <MSLofar/FailedTileInfo.h>
 #include <CoInterface/CorrelatedData.h>
 #include <CoInterface/LTAFeedback.h>
@@ -48,18 +47,18 @@ namespace LOFAR
 
     MSWriterCorrelated::MSWriterCorrelated (const std::string &logPrefix, const std::string &msName, const Parset &parset, unsigned subbandIndex)
       :
-      MSWriterFile(
-        (makeMeasurementSet(logPrefix, msName, parset, subbandIndex),
-         str(format("%s/table.f0data") % msName))),
+      MSWriterFile(str(format("%s/table.f0data") % msName)),
       itsLogPrefix(logPrefix),
       itsMSname(msName),
-      itsParset(parset)
+      itsParset(parset),
+      itsSubbandIndex(subbandIndex)
     {
       // Add file-specific processing feedback
       LTAFeedback fb(itsParset.settings);
-      itsConfiguration.adoptCollection(fb.correlatedFeedback(subbandIndex));
-      itsConfigurationPrefix = fb.correlatedPrefix(subbandIndex);
+      itsConfiguration.adoptCollection(fb.correlatedFeedback(itsSubbandIndex));
+      itsConfigurationPrefix = fb.correlatedPrefix(itsSubbandIndex);
 
+      // Create Sequence file
       if (LofarStManVersion > 1) {
         string seqfilename = str(format("%s/table.f0seqnr") % msName);
 
@@ -69,23 +68,19 @@ namespace LOFAR
           LOG_WARN_STR(itsLogPrefix << "Could not open sequence numbers file " << seqfilename);
         }
       }
+    }
 
-#if 0
-      // derive baseline names
-      std::vector<std::string> stationNames = parset.mergedStationNames();
-      std::vector<std::string> baselineNames(parset.nrBaselines());
-      unsigned nrStations = stationNames.size();
 
-      // order of baselines as station indices:
-      // 0-0, 1-0, 1-1, 2-0, 2-1, 2-2 ... (see RTCP/CNProc/Correlator.cc)
+    void MSWriterCorrelated::createMetaData()
+    {
+      // Creaate MeasurementSet
+#if defined HAVE_AIPSPP
+      MeasurementSetFormat myFormat(itsParset, 512);
 
-      unsigned bl = 0;
+      myFormat.addSubband(itsMSname, itsSubbandIndex);
 
-      for(unsigned s1 = 0; s1 < nrStations; s1++)
-        for(unsigned s2 = 0; s2 <= s1; s2++)
-          //bl = s1 * (s1 + 1) / 2 + stat2 ;
-          baselineNames[bl++] = str(format("%s_%s") % stationNames[s1] % stationNames[s2]);
-#endif
+      LOG_DEBUG_STR(itsLogPrefix << "MeasurementSet created");
+#endif // defined HAVE_AIPSPP
     }
 
 
@@ -94,18 +89,6 @@ namespace LOFAR
     }
 
 
-    void MSWriterCorrelated::makeMeasurementSet(const std::string &logPrefix, const std::string &msName, const Parset &parset, unsigned subbandIndex)
-    {
-#if defined HAVE_AIPSPP
-      MeasurementSetFormat myFormat(parset, 512);
-
-      myFormat.addSubband(msName, subbandIndex);
-
-      LOG_DEBUG_STR(logPrefix << "MeasurementSet created");
-#endif // defined HAVE_AIPSPP
-    }
-
-
     void MSWriterCorrelated::write(StreamableData *data)
     {
       CorrelatedData *cdata = dynamic_cast<CorrelatedData*>(data);

RTCP/Cobalt/OutputProc/src/MSWriterCorrelated.h

@@ -43,6 +43,8 @@ namespace LOFAR
       MSWriterCorrelated(const std::string &logPrefix, const std::string &msName, const Parset &parset, unsigned subbandIndex);
       ~MSWriterCorrelated();
 
+      virtual void createMetaData();
+
       virtual void write(StreamableData *data);
 
       virtual void augment(const FinalMetaData &finalMetaData);
@@ -51,11 +53,9 @@ namespace LOFAR
       const std::string itsLogPrefix;
       const std::string itsMSname;
       const Parset &itsParset;
+      const unsigned itsSubbandIndex;
 
       SmartPtr<FileStream>             itsSequenceNumbersFile;
-
-    private:
-      void makeMeasurementSet(const std::string &logPrefix, const std::string &msName, const Parset &parset, unsigned subbandIndex);
     };
 
 

RTCP/Cobalt/OutputProc/src/MSWriterDAL.cc

@@ -83,20 +83,41 @@ namespace LOFAR
      const Parset &parset,
      unsigned fileno)
       :
-      MSWriterFile(forceextension(string(filename),".raw")),
+      MSWriterFile(forceextension(string(filename), ".raw")),
+      itsFilename(filename),
       itsParset(parset),
       itsNextSeqNr(0),
       itsFileNr(fileno)
     {
+      itsNrExpectedBlocks = itsParset.settings.nrBlocks();
+
+      const struct ObservationSettings::BeamFormer::File &f = itsParset.settings.beamFormer.files[itsFileNr];
+
+      const struct ObservationSettings::BeamFormer::StokesSettings &stokesSet =
+        f.coherent ? itsParset.settings.beamFormer.coherentSettings
+                   : itsParset.settings.beamFormer.incoherentSettings;
+
+      // All subbands in the SAP that we store in this file.
+      // We could have multiple SAPs and/or have split up the subbands over multiple files (parts).
+      unsigned nrSubbands = f.lastSubbandIdx - f.firstSubbandIdx;
+
+      itsNrChannels = stokesSet.nrChannels * nrSubbands; 
+      itsNrSamples = itsParset.settings.blockSize /
+                     stokesSet.nrChannels / stokesSet.timeIntegrationFactor;
+
+      itsBlockSize = itsNrSamples * itsNrChannels;
+
       // Add file-specific processing feedback
       LTAFeedback fb(itsParset.settings);
       itsConfiguration.adoptCollection(fb.beamFormedFeedback(itsFileNr));
       itsConfigurationPrefix = fb.beamFormedPrefix(itsFileNr);
+    }
 
-      itsNrExpectedBlocks = itsParset.settings.nrBlocks();
-
-      string h5filename = forceextension(string(filename),".h5");
-      string rawfilename = forceextension(string(filename),".raw");
+    template <typename T,unsigned DIM>
+    void MSWriterDAL<T,DIM>::createMetaData()
+    {
+      string h5filename = forceextension(itsFilename, ".h5");
+      string rawfilename = forceextension(itsFilename, ".raw");
 
       ScopedLock sl(HDF5Mutex);
 
@@ -105,15 +126,15 @@ namespace LOFAR
       H5Eset_auto_stack(H5E_DEFAULT, my_hdf5_error_handler, NULL);
 #endif
 
-      const struct ObservationSettings::BeamFormer::File &f = parset.settings.beamFormer.files[fileno];
+      const struct ObservationSettings::BeamFormer::File &f = itsParset.settings.beamFormer.files[itsFileNr];
 
       const unsigned sapNr = f.sapNr;
       const unsigned beamNr = f.tabNr;
       const unsigned stokesNr = f.stokesNr;
 
       const struct ObservationSettings::BeamFormer::StokesSettings &stokesSet =
-        f.coherent ? parset.settings.beamFormer.coherentSettings
-                   : parset.settings.beamFormer.incoherentSettings;
+        f.coherent ? itsParset.settings.beamFormer.coherentSettings
+                   : itsParset.settings.beamFormer.incoherentSettings;
 
       //*******************************
 
@@ -122,12 +143,6 @@ namespace LOFAR
       unsigned firstSubbandIdx = f.firstSubbandIdx;
       unsigned nrSubbands = f.lastSubbandIdx - f.firstSubbandIdx;
 
-      itsNrChannels = stokesSet.nrChannels * nrSubbands; 
-      itsNrSamples = parset.settings.blockSize /
-                     stokesSet.nrChannels / stokesSet.timeIntegrationFactor;
-
-      itsBlockSize = itsNrSamples * itsNrChannels;
-
       //*******************************
 
       vector<string> stokesVars;
@@ -164,30 +179,30 @@ namespace LOFAR
         return;
       }
 
-      LOG_DEBUG_STR("MSWriterDAL: opening " << filename);
+      LOG_DEBUG_STR("MSWriterDAL: opening " << h5filename);
 
       // create the top structure
       BF_File file(h5filename, BF_File::CREATE);
 
       // Common Attributes
-      writeCommonLofarAttributes(file, parset);
+      writeCommonLofarAttributes(file, itsParset);
 
       // BF_File specific root group parameters
-      file.createOfflineOnline().value = parset.settings.realTime ? "Online" : "Offline";
+      file.createOfflineOnline().value = itsParset.settings.realTime ? "Online" : "Offline";
       file.BFFormat().value = "TAB";
       file.BFVersion().value = str(format("Cobalt/OutputProc %s r%s using DAL %s and HDF5 %s") % OutputProcVersion::getVersion() % OutputProcVersion::getRevision() % dal::version().to_string() % dal::version_hdf5().to_string());
 
-      file.totalIntegrationTime().value = itsNrExpectedBlocks * parset.settings.blockDuration();
+      file.totalIntegrationTime().value = itsNrExpectedBlocks * itsParset.settings.blockDuration();
       file.totalIntegrationTimeUnit().value = "s";
 
       //file.subArrayPointingDiameter().value = 0.0;
       //file.subArrayPointingDiameterUnit().value = "arcmin";
-      file.bandwidth().value = parset.settings.subbands.size() * parset.settings.subbandWidth() / 1e6;
+      file.bandwidth().value = itsParset.settings.subbands.size() * itsParset.settings.subbandWidth() / 1e6;
       file.bandwidthUnit().value = "MHz";
       //file.beamDiameter()            .value = 0.0;
       //file.beamDiameterUnit()          .value = "arcmin";
 
-      file.observationNofSubArrayPointings().value = parset.settings.SAPs.size();
+      file.observationNofSubArrayPointings().value = itsParset.settings.SAPs.size();
       file.nofSubArrayPointings().value = 1;
 
       // SysLog group -- empty for now
@@ -198,30 +213,30 @@ namespace LOFAR
       sap.create();
       sap.groupType().value = "SubArrayPointing";
 
-      sap.expTimeStartUTC().value = toUTC(parset.settings.startTime);
-      sap.expTimeStartMJD().value = toMJD(parset.settings.startTime);
+      sap.expTimeStartUTC().value = toUTC(itsParset.settings.startTime);
+      sap.expTimeStartMJD().value = toMJD(itsParset.settings.startTime);
 
-      sap.expTimeEndUTC().value = toUTC(parset.getRealStopTime());
-      sap.expTimeEndMJD().value = toMJD(parset.getRealStopTime());
+      sap.expTimeEndUTC().value = toUTC(itsParset.getRealStopTime());
+      sap.expTimeEndMJD().value = toMJD(itsParset.getRealStopTime());
 
-      // TODO: fix the system to use the parset.beamDuration(sapNr), but OLAP
+      // TODO: fix the system to use the itsParset.beamDuration(sapNr), but OLAP
       // does not work that way yet (beamDuration is currently unsupported).
-      sap.totalIntegrationTime().value = itsNrExpectedBlocks * parset.settings.blockDuration();
+      sap.totalIntegrationTime().value = itsNrExpectedBlocks * itsParset.settings.blockDuration();
       sap.totalIntegrationTimeUnit().value = "s";
 
       // TODO: non-J2000 pointings.
       // Idem for TABs: now we subtract absolute angles to store TAB offsets. Also see TODO below.
-      if ( parset.settings.SAPs[sapNr].direction.type != "J2000" ) {
+      if ( itsParset.settings.SAPs[sapNr].direction.type != "J2000" ) {
         LOG_WARN("HDF5 writer does not record positions of non-J2000 observations yet.");
       }
 
-      const struct ObservationSettings::Direction &beamDir = parset.settings.SAPs[sapNr].direction;
+      const struct ObservationSettings::Direction &beamDir = itsParset.settings.SAPs[sapNr].direction;
       sap.pointRA().value = beamDir.angle1 * 180.0 / M_PI;
       sap.pointRAUnit().value = "deg";
       sap.pointDEC().value = beamDir.angle2 * 180.0 / M_PI;
       sap.pointDECUnit().value = "deg";
 
-      sap.observationNofBeams().value = parset.settings.beamFormer.SAPs[sapNr].TABs.size();
+      sap.observationNofBeams().value = itsParset.settings.beamFormer.SAPs[sapNr].TABs.size();
       sap.nofBeams().value = 1;
 
       BF_ProcessingHistory sapHistory = sap.processHistory();
@@ -229,10 +244,10 @@ namespace LOFAR
       sapHistory.groupType().value = "ProcessingHistory";
 
       Attribute<string> sapObservationParset(sapHistory, "OBSERVATION_PARSET");
-      string parsetAsString;
-      parset.writeBuffer(parsetAsString);
+      string itsParsetAsString;
+      itsParset.writeBuffer(itsParsetAsString);
 
-      sapObservationParset.value = parsetAsString;
+      sapObservationParset.value = itsParsetAsString;
 
       // information about the pencil beam
 
@@ -240,20 +255,20 @@ namespace LOFAR
       beam.create();
       beam.groupType().value = "Beam";
 
-      if (parset.settings.beamFormer.doFlysEye) {
+      if (itsParset.settings.beamFormer.doFlysEye) {
         beam.nofStations().value = 1;
-        beam.stationsList().value = vector<string>(1, parset.settings.antennaFields[beamNr].name);
+        beam.stationsList().value = vector<string>(1, itsParset.settings.antennaFields[beamNr].name);
       } else {
-        beam.nofStations().value = parset.settings.antennaFields.size();
-        beam.stationsList().value = parset.allStationNames();
+        beam.nofStations().value = itsParset.settings.antennaFields.size();
+        beam.stationsList().value = itsParset.allStationNames();
       }
 
-      const vector<string> beamtargets(1, parset.settings.SAPs[sapNr].target);
+      const vector<string> beamtargets(1, itsParset.settings.SAPs[sapNr].target);
 
       beam.targets().value = beamtargets;
-      beam.tracking().value = parset.settings.SAPs[sapNr].direction.type;
+      beam.tracking().value = itsParset.settings.SAPs[sapNr].direction.type;
 
-      const struct ObservationSettings::Direction &tabDir = parset.settings.beamFormer.SAPs[sapNr].TABs[beamNr].direction;
+      const struct ObservationSettings::Direction &tabDir = itsParset.settings.beamFormer.SAPs[sapNr].TABs[beamNr].direction;
       beam.pointRA().value = tabDir.angle1 * 180.0 / M_PI;
       beam.pointRAUnit().value = "deg";
       beam.pointDEC().value = tabDir.angle2 * 180.0 / M_PI;
@@ -264,7 +279,7 @@ namespace LOFAR
       beam.pointOffsetDECUnit().value = "deg";
 
 
-      beam.subbandWidth().value = parset.settings.subbandWidth();
+      beam.subbandWidth().value = itsParset.settings.subbandWidth();
       beam.subbandWidthUnit().value = "Hz";
 
       beam.beamDiameterRA().value = 0;
@@ -273,20 +288,20 @@ namespace LOFAR
       beam.beamDiameterDECUnit().value = "arcmin";
 
       beam.nofSamples().value = itsNrSamples * itsNrExpectedBlocks;
-      beam.samplingRate().value = parset.settings.subbandWidth() / stokesSet.nrChannels / stokesSet.timeIntegrationFactor;
+      beam.samplingRate().value = itsParset.settings.subbandWidth() / stokesSet.nrChannels / stokesSet.timeIntegrationFactor;
       beam.samplingRateUnit().value = "Hz";
-      beam.samplingTime().value = parset.settings.sampleDuration() * stokesSet.nrChannels * stokesSet.timeIntegrationFactor;
+      beam.samplingTime().value = itsParset.settings.sampleDuration() * stokesSet.nrChannels * stokesSet.timeIntegrationFactor;
       beam.samplingTimeUnit().value = "s";
 
       beam.channelsPerSubband().value = stokesSet.nrChannels;
-      const double channelBandwidth = parset.settings.subbandWidth() / stokesSet.nrChannels;
+      const double channelBandwidth = itsParset.settings.subbandWidth() / stokesSet.nrChannels;
       beam.channelWidth().value = channelBandwidth;
       beam.channelWidthUnit().value = "Hz";
 
       // First, init tmp vector for the whole obs, regardless which SAP and subbands (parts) this file contains.
-      vector<double> subbandCenterFrequencies(parset.settings.subbands.size());
+      vector<double> subbandCenterFrequencies(itsParset.settings.subbands.size());
       for (unsigned sb = 0; sb < subbandCenterFrequencies.size(); ++sb) {
-        subbandCenterFrequencies[sb] = parset.settings.subbands[sb].centralFrequency;
+        subbandCenterFrequencies[sb] = itsParset.settings.subbands[sb].centralFrequency;
       }
 
       vector<double> beamCenterFrequencies(nrSubbands, 0.0);
@@ -296,12 +311,12 @@ namespace LOFAR
 
       const double beamCenterFrequencySum = accumulate(beamCenterFrequencies.begin(), beamCenterFrequencies.end(), 0.0);
       const double frequencyOffsetPPF = stokesSet.nrChannels > 1 ? // See getFrequencyOffsetPPF() for why/how.
-                                        getFrequencyOffsetPPF(parset.settings.subbandWidth(), stokesSet.nrChannels) :
+                                        getFrequencyOffsetPPF(itsParset.settings.subbandWidth(), stokesSet.nrChannels) :
                                         0.0;
       beam.beamFrequencyCenter().value = (beamCenterFrequencySum / nrSubbands - frequencyOffsetPPF) / 1e6;
       beam.beamFrequencyCenterUnit().value = "MHz";
 
-      const double DM = parset.settings.corrections.dedisperse ? parset.settings.beamFormer.SAPs[sapNr].TABs[beamNr].dispersionMeasure : 0.0;
+      const double DM = itsParset.settings.corrections.dedisperse ? itsParset.settings.beamFormer.SAPs[sapNr].TABs[beamNr].dispersionMeasure : 0.0;
 
       beam.foldedData().value = false;
       beam.foldPeriod().value = 0.0;
@@ -331,17 +346,17 @@ namespace LOFAR
 
       Attribute<string> beamObservationParset(beamHistory, "OBSERVATION_PARSET");
 
-      beamObservationParset.value = parsetAsString;
+      beamObservationParset.value = itsParsetAsString;
 
       CoordinatesGroup coordinates = beam.coordinates();
       coordinates.create();
       coordinates.groupType().value = "Coordinates";
 
-      coordinates.refLocationValue().value = parset.settings.delayCompensation.referencePhaseCenter;
+      coordinates.refLocationValue().value = itsParset.settings.delayCompensation.referencePhaseCenter;
       coordinates.refLocationUnit().value = vector<string>(3, "m");
       coordinates.refLocationFrame().value = "ITRF";
 
-      coordinates.refTimeValue().value = toMJD(parset.settings.startTime);
+      coordinates.refTimeValue().value = toMJD(itsParset.settings.startTime);
       coordinates.refTimeUnit().value = "d";
       coordinates.refTimeFrame().value = "MJD";
 
@@ -373,7 +388,7 @@ namespace LOFAR
 
       timeCoordinate.get()->referenceValue().value = 0;
       timeCoordinate.get()->referencePixel().value = 0;
-      timeCoordinate.get()->increment().value = parset.sampleDuration() * stokesSet.nrChannels * stokesSet.timeIntegrationFactor;
+      timeCoordinate.get()->increment().value = itsParset.sampleDuration() * stokesSet.nrChannels * stokesSet.timeIntegrationFactor;
       timeCoordinate.get()->pc().value = unitvector;
 
       timeCoordinate.get()->axisValuesPixel().value = vector<unsigned>(1, 0); // not used
@@ -402,7 +417,7 @@ namespace LOFAR
       vector<double> spectralWorld;
 
       for (unsigned sb = 0; sb < nrSubbands; sb++) {
-        const double subbandBeginFreq = parset.channel0Frequency( firstSubbandIdx + sb, stokesSet.nrChannels );
+        const double subbandBeginFreq = itsParset.channel0Frequency( firstSubbandIdx + sb, stokesSet.nrChannels );
 
         // NOTE: channel 0 will be wrongly annotated if nrChannels > 1, because it is a combination of the
         // highest and the lowest frequencies (half a channel each).

RTCP/Cobalt/OutputProc/src/MSWriterDAL.h

@@ -44,13 +44,18 @@ namespace LOFAR
     public:
       MSWriterDAL(const std::string &filename, const Parset &parset, unsigned fileno);
       ~MSWriterDAL();
+
+      virtual void createMetaData();
+
       virtual void write(StreamableData *data);
+
     private:
+      const std::string itsFilename;
       const Parset &itsParset;
       unsigned itsNrChannels;
       unsigned itsNrSamples;
       unsigned itsNextSeqNr;
-      unsigned itsFileNr;
+      const unsigned itsFileNr;
       unsigned itsBlockSize; // the size of StreamableData::samples, in T
     };
   }

RTCP/Cobalt/OutputProc/src/MSWriterFile.cc

@@ -21,19 +21,64 @@
 #include <lofar_config.h>
 
 #include "MSWriterFile.h"
+#include <Common/Thread/Mutex.h>
+#include <Common/SystemUtil.h>
 
 #include <sys/types.h>
+#include <sys/stat.h>
 #include <fcntl.h>
+#include <cerrno>
+#include <boost/algorithm/string.hpp>
 
 namespace LOFAR
 {
   namespace Cobalt
   {
+    static Mutex makeDirMutex;
+
+    static void makeDir(const string &dirname, const string &logPrefix)
+    {
+      ScopedLock scopedLock(makeDirMutex);
+      struct stat s;
+
+      if (stat(dirname.c_str(), &s) == 0) {
+        // path already exists
+        if ((s.st_mode & S_IFMT) != S_IFDIR) {
+          LOG_WARN_STR(logPrefix << "Not a directory: " << dirname);
+        }
+      } else if (errno == ENOENT) {
+        // create directory
+        LOG_DEBUG_STR(logPrefix << "Creating directory " << dirname);
+
+        if (mkdir(dirname.c_str(), 0777) != 0 && errno != EEXIST) {
+          THROW_SYSCALL(string("mkdir ") + dirname);
+        }
+      } else {
+        // something else went wrong
+        THROW_SYSCALL(string("stat ") + dirname);
+      }
+    }
+
+    /* create a directory as well as all its parent directories */
+    static void recursiveMakeDir(const string &dirname, const string &logPrefix)
+    {
+      using namespace boost;
+
+      string curdir;
+      vector<string> splitName;
+
+      boost::split(splitName, dirname, boost::is_any_of("/"));
+
+      for (unsigned i = 0; i < splitName.size(); i++) {
+        curdir += splitName[i] + '/';
+        makeDir(curdir, logPrefix);
+      }
+    }
 
 
     MSWriterFile::MSWriterFile (const std::string &msName)
       :
-      itsFile(msName, O_RDWR | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)
+      itsFile((recursiveMakeDir(dirname(msName), ""), msName), O_RDWR | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)
     {
     }
 

RTCP/Cobalt/OutputProc/src/MSWriterFile.h

@@ -36,6 +36,11 @@ namespace LOFAR
     class MSWriterFile : public MSWriter
     {
     public:
+      /*
+       * Write data to the provided file name.
+       *
+       * Any parent directories are automatically created.
+       */
       MSWriterFile(const std::string &msName);
       ~MSWriterFile();
 

RTCP/Cobalt/OutputProc/src/OutputThread.cc

@@ -23,17 +23,11 @@
 
 #include "OutputThread.h"
 
-#include <cerrno>
-#include <ctime>
-#include <sys/types.h>
-#include <sys/stat.h>
 #include <unistd.h>
 #include <iomanip>
 #include <boost/lexical_cast.hpp>
 #include <boost/format.hpp>
-#include <boost/algorithm/string.hpp>
 
-#include <Common/StringUtil.h>
 #include <Common/SystemCallException.h>
 #include <Common/Thread/Mutex.h>
 #include <Common/Thread/Cancellation.h>
@@ -57,53 +51,11 @@ namespace LOFAR
   namespace Cobalt
   {
 
-    static Mutex makeDirMutex;
     static Mutex casacoreMutex;
 
     using namespace std;
     using boost::lexical_cast;
 
-    static void makeDir(const string &dirname, const string &logPrefix)
-    {
-      ScopedLock scopedLock(makeDirMutex);
-      struct stat s;
-
-      if (stat(dirname.c_str(), &s) == 0) {
-        // path already exists
-        if ((s.st_mode & S_IFMT) != S_IFDIR) {
-          LOG_WARN_STR(logPrefix << "Not a directory: " << dirname);
-        }
-      } else if (errno == ENOENT) {
-        // create directory
-        LOG_DEBUG_STR(logPrefix << "Creating directory " << dirname);
-
-        if (mkdir(dirname.c_str(), 0777) != 0 && errno != EEXIST) {
-          THROW_SYSCALL(string("mkdir ") + dirname);
-        }
-      } else {
-        // something else went wrong
-        THROW_SYSCALL(string("stat ") + dirname);
-      }
-    }
-
-
-    /* create a directory as well as all its parent directories */
-    static void recursiveMakeDir(const string &dirname, const string &logPrefix)
-    {
-      using namespace boost;
-
-      string curdir;
-      vector<string> splitName;
-
-      boost::split(splitName, dirname, boost::is_any_of("/"));
-
-      for (unsigned i = 0; i < splitName.size(); i++) {
-        curdir += splitName[i] + '/';
-        makeDir(curdir, logPrefix);
-      }
-    }
-
-
     template<typename T> OutputThread<T>::OutputThread(const Parset &parset,
           unsigned streamNr, Pool<T> &outputPool,
           RTmetadata &mdLogger, const std::string &mdKeyPrefix,
@@ -209,6 +161,19 @@ namespace LOFAR
     }
 
 
+    template<typename T> void OutputThread<T>::createMetaData()
+    {
+      try {
+        // augment the data product
+        ASSERT(itsWriter.get());
+
+        itsWriter->createMetaData();
+      } catch (Exception &ex) {
+        LOG_ERROR_STR(itsLogPrefix << "Could not create meta data: " << ex);
+      }
+    }
+
+
     template<typename T> void OutputThread<T>::augment( const FinalMetaData &finalMetaData )
     {
       try {
@@ -241,8 +206,18 @@ namespace LOFAR
       LOG_DEBUG_STR(itsLogPrefix << "process() entered");
 
       createMS();
-      doWork();
-      cleanUp();
+
+#     pragma omp parallel sections num_threads(2)
+      {
+#       pragma omp section
+        {
+          doWork();
+          cleanUp();
+        }
+
+#       pragma omp section
+        createMetaData();
+      }
     }
 
     // Make required instantiations
@@ -282,7 +257,6 @@ namespace LOFAR
 
       try
       {
-        recursiveMakeDir(directoryName, itsLogPrefix);
         LOG_INFO_STR(itsLogPrefix << "Writing to " << path);
 
         itsWriter = new MSWriterCorrelated(itsLogPrefix, path, itsParset, itsStreamNr);
@@ -347,7 +321,6 @@ namespace LOFAR
 
       try
       {
-        recursiveMakeDir(directoryName, itsLogPrefix);
         LOG_INFO_STR(itsLogPrefix << "Writing to " << path);
 
 #ifdef HAVE_DAL

RTCP/Cobalt/OutputProc/src/OutputThread.h

@@ -69,6 +69,9 @@ namespace LOFAR
       // Wrap-up the writing.
       void           cleanUp() const;
 
+      // Create the metadata files in parallel to writing the data
+      void           createMetaData();
+
       // Add FinalMetaData to the data container.
       void           augment(const FinalMetaData &finalMetaData);