Task #4820: Replaced Common/Thread model with OpenMP

RTCP/Cobalt/OutputProc/src/InputThread.cc

@@ -49,20 +49,7 @@ namespace LOFAR
     }
 
 
-    void InputThread::start()
-    {
-      itsThread = new Thread(this, &InputThread::mainLoop, itsLogPrefix);
-    }
-
-
-    void InputThread::cancel()
-    {
-      if (itsThread)
-        itsThread->cancel();
-    }
-
-
-    void InputThread::mainLoop()
+    void InputThread::process()
     {
       try {
         LOG_INFO_STR(itsLogPrefix << "Creating connection from " << itsInputDescriptor << "..." );

RTCP/Cobalt/OutputProc/src/InputThread.h

@@ -25,7 +25,6 @@
 
 #include <string>
 
-#include <Common/Thread/Thread.h>
 #include <Common/Thread/Queue.h>
 #include <CoInterface/OutputTypes.h>
 #include <CoInterface/Parset.h>
@@ -49,15 +48,11 @@ namespace LOFAR
                   Queue<SmartPtr<StreamableData> > &receiveQueue,
                   const std::string &logPrefix);
 
-      void                             start();
-      void                             cancel();
+      void                             process();
 
     private:
-      void                             mainLoop();
-
       const std::string itsLogPrefix, itsInputDescriptor;
       Queue<SmartPtr<StreamableData> > &itsFreeQueue, &itsReceiveQueue;
-      SmartPtr<Thread>                 itsThread;
       const double itsDeadline;
     };
 

RTCP/Cobalt/OutputProc/src/OutputThread.cc

@@ -30,6 +30,7 @@
 #include <unistd.h>
 #include <iomanip>
 #include <boost/format.hpp>
+#include <boost/algorithm/string.hpp>
 
 #include <Common/StringUtil.h>
 #include <Common/SystemCallException.h>
@@ -88,7 +89,7 @@ namespace LOFAR
       string curdir;
       vector<string> splitName;
 
-      split(splitName, dirname, is_any_of("/"));
+      boost::split(splitName, dirname, boost::is_any_of("/"));
 
       for (unsigned i = 0; i < splitName.size(); i++) {
         curdir += splitName[i] + '/';
@@ -114,12 +115,6 @@ namespace LOFAR
     }
 
 
-    void OutputThread::start()
-    {
-      itsThread = new Thread(this, &OutputThread::mainLoop, itsLogPrefix);
-    }
-
-
     void OutputThread::createMS()
     {
       // even the HDF5 writer accesses casacore, to perform conversions
@@ -259,11 +254,6 @@ namespace LOFAR
 
     void OutputThread::augment( const FinalMetaData &finalMetaData )
     {
-      // wait for writer thread to finish, so we'll have an itsWriter
-      ASSERT(itsThread.get());
-
-      itsThread = 0;
-
       // augment the data product
       ASSERT(itsWriter.get());
 
@@ -271,9 +261,9 @@ namespace LOFAR
     }
 
 
-    void OutputThread::mainLoop()
+    void OutputThread::process()
     {
-      LOG_DEBUG_STR(itsLogPrefix << "OutputThread::mainLoop() entered");
+      LOG_DEBUG_STR(itsLogPrefix << "OutputThread::process() entered");
 
       try {
         createMS();

RTCP/Cobalt/OutputProc/src/OutputThread.h

@@ -27,7 +27,6 @@
 #include <vector>
 
 #include <Common/Thread/Queue.h>
-#include <Common/Thread/Thread.h>
 #include <Stream/FileStream.h>
 #include <CoInterface/OutputTypes.h>
 #include <CoInterface/SmartPtr.h>
@@ -46,7 +45,7 @@ namespace LOFAR
     public:
       OutputThread(const Parset &, OutputType, unsigned streamNr, Queue<SmartPtr<StreamableData> > &freeQueue, Queue<SmartPtr<StreamableData> > &receiveQueue, const std::string &logPrefix, const std::string &targetDirectory = "");
 
-      void                             start();
+      void           process();
 
       // needed in createHeaders.cc
       void           createMS();
@@ -59,7 +58,6 @@ namespace LOFAR
     private:
       void                             checkForDroppedData(StreamableData *);
       void                             doWork();
-      void                             mainLoop();
 
       const Parset                     &itsParset;
       const OutputType itsOutputType;
@@ -73,7 +71,6 @@ namespace LOFAR
       unsigned itsNrExpectedBlocks;
       unsigned itsNextSequenceNumber;
       SmartPtr<MSWriter>               itsWriter;
-      SmartPtr<Thread>                 itsThread;
     };
 
 

RTCP/Cobalt/OutputProc/src/SubbandWriter.cc

@@ -33,22 +33,27 @@ namespace LOFAR
     SubbandWriter::SubbandWriter(const Parset &parset, OutputType outputType, unsigned streamNr, const std::string &logPrefix)
     {
       itsInputThread = new InputThread(parset, outputType, streamNr, itsFreeQueue, itsReceiveQueue, logPrefix);
-      itsInputThread->start();
-
-      try 
-      {
-        itsOutputThread = new OutputThread(parset, outputType, streamNr, itsFreeQueue, itsReceiveQueue, logPrefix);
-        itsOutputThread->start();
-      } 
-      catch (...) 
-      {
-        itsInputThread->cancel();
-        throw;
-      }
+      itsOutputThread = new OutputThread(parset, outputType, streamNr, itsFreeQueue, itsReceiveQueue, logPrefix);
 
       for (unsigned i = 0; i < maxReceiveQueueSize; i++)
         itsFreeQueue.append(newStreamableData(parset, outputType, streamNr));
+    }
 
+    
+    void SubbandWriter::process()
+    {
+#     pragma omp parallel sections num_threads(2)
+      {
+#       pragma omp section
+        {
+          itsInputThread->process();
+        }
+
+#       pragma omp section
+        {
+          itsOutputThread->process();
+        }
+      }
     }
 
     void SubbandWriter::augment( const FinalMetaData &finalMetaData )

RTCP/Cobalt/OutputProc/src/SubbandWriter.h

@@ -45,6 +45,8 @@ namespace LOFAR
                     unsigned streamNr,
                     const std::string &logPrefix);
 
+      void process();
+
       void augment(const FinalMetaData &finalMetaData);
 
       ParameterSet feedbackLTA() const;

RTCP/Cobalt/OutputProc/src/outputProc.cc

@@ -24,6 +24,7 @@
 #include <cstdlib>
 #include <cstdio>
 #include <cstring>
+#include <omp.h>
 
 #include <string>
 #include <vector>
@@ -139,6 +140,10 @@ void process(Stream &controlStream, size_t myRank)
 
     vector<SmartPtr<SubbandWriter> > subbandWriters;
 
+    /*
+     * Construct writers
+     */
+
     // Process correlated data
     if (parset.settings.correlator.enabled) {
       for (size_t fileIdx = 0; fileIdx < parset.settings.correlator.files.size(); ++fileIdx)
@@ -169,6 +174,16 @@ void process(Stream &controlStream, size_t myRank)
       }
     }
 
+    /*
+     * PROCESS
+     */
+
+#   pragma omp parallel for num_threads(subbandWriters.size())
+    for (int i = 0; i < (int)subbandWriters.size(); ++i)
+    {
+      subbandWriters[i]->process();
+    }
+
     /*
      * FINAL META DATA
      */
@@ -194,6 +209,8 @@ int main(int argc, char *argv[])
   setvbuf(stdout, stdoutbuf, _IOLBF, sizeof stdoutbuf);
   setvbuf(stderr, stderrbuf, _IOLBF, sizeof stderrbuf);
 
+  omp_set_nested(true);
+
   LOG_DEBUG_STR("Started: " << argv[0] << ' ' << argv[1] << ' ' << argv[2]);
 
   int observationID = boost::lexical_cast<int>(argv[1]);