Changeset 10951

Timestamp:

06/09/11 15:44:01 (13 years ago)

Author:

lyard

Message:

moved various processed from secondary thread to main thread

File:

• trunk/FACT++/src/dataLogger.cc (modified) (26 diffs)

trunk/FACT++/src/dataLogger.cc

@@ -63 +63 @@
 
 #include <fstream>
+#include <mutex>
 
 #include <boost/bind.hpp>
@@ -135 +136 @@
 
     }
-    ///copy operator
-    /*
-    void operator = (const RunNumberType& other)
-    {
-#ifdef RUN_LOGS
-        logFile = other.logFile;
-        logName = other.logName;
-#endif
-        reportFile = other.reportFile;
-        reportName = other.reportName;
-        runNumber = other.runNumber;
-        time = other.time;
-#ifdef HAVE_FITS
-        runFitsFile = other.runFitsFile;
-#endif
-}*/
-
-    ///copy constructor
-    RunNumberType(const RunNumberType& other)
-    {
-#ifdef RUN_LOGS
-        logFile = other.logFile;
-        logName = other.logName;
-#endif
-        reportFile = other.reportFile;
-        reportName = other.reportName;
-        runNumber = other.runNumber;
-        time = other.time;
-#ifdef HAVE_FITS
-        runFitsFile = other.runFitsFile;
-#endif
-    }
 
     void addServiceToOpenedFits(const string& fileName, const string& serviceName)
@@ -198 +167 @@
     ///the current run number used by this subscription
     uint32_t runNumber;
-    ///copy operator
-    /*
-    void operator = (const SubscriptionType& other)
-    {
-#ifdef HAVE_FITS
-        nightlyFile = other.nightlyFile;
-        runFile = other.runFile;
-#endif
-        dimInfo = other.dimInfo;
-        server = other.server;
-        service = other.service;
-        fConv = other.fConv;
-        runNumber = other.runNumber;
-        }*/
-
-/*    ///copy constructor
-    SubscriptionType(const SubscriptionType& other)
-    {
-#ifdef HAVE_FITS
-        nightlyFile = other.nightlyFile;
-        runFile = other.runFile;
-#endif
-        dimInfo = other.dimInfo;
-        server = other.server;
-        service = other.service;
-        fConv = other.fConv;
-        runNumber = other.runNumber;
-    }*/
+    ///time of the latest received event
+    Time lastReceivedEvent;
+    ///whether or not the fits buffer was allocated already
+    bool fitsBufferAllocated;
+
     ///Dim info constructor
     SubscriptionType(DimStampedInfo* info=NULL)
@@ -232 +178 @@
         fConv = shared_ptr<Converter>();
         runNumber = 0;
-    }
-    ///default constructor
-/*    SubscriptionType()
-    {
-        dimInfo = shared_ptr<DimStampedInfo>();
-        fConv = shared_ptr<Converter>();
-        runNumber = 0;
-    }*/
+        lastReceivedEvent = Time::None;
+        fitsBufferAllocated = false;
+    }
+
     ///default destructor
     ~SubscriptionType()
@@ -293 +235 @@
     ///full name of the nightly report file
     string fFullNightlyReportFileName;
-
+    ///variables for computing statistics
+    DataLoggerStats fStatVar;
+    ///variable to track when the statistic were last calculated
+    double fPreviousStatsUpdateTime;
+    double fPreviousOldRunNumberCheck;
+    ///mutex to make sure that the Stats are not accessed while updating
+    mutex fStatsMutex;
+    ///boolean to know whether we should close and reopen daily files or not
+    bool fDailyFileDayChangedAlready;
 public:
     /***************************************************
@@ -556 +506 @@
     list[service].server  = server;
     list[service].service = service;
-
+    fNumSubAndFitsData.numSubscriptions++;
     if (fDebugIsOn)
         Debug("Added subscription to " + server + "/" + service);
@@ -571 +521 @@
     if (isCmd)
         return;
-
-    fServiceSubscriptions[server].erase(service);
+    if (fServiceSubscriptions[server].erase(service) != 1)
+    {
+        ostringstream str;
+        str << "Subscription " << server << "/" << service << "could not be removed as it is not present";
+        Error(str.str());
+        return;
+    }
+    fNumSubAndFitsData.numSubscriptions--;
     if (fDebugIsOn)
     {
@@ -585 +541 @@
 void DataLogger::RemoveAllServices(const string& server)
 {
+    fNumSubAndFitsData.numSubscriptions -= fServiceSubscriptions[server].size();
     fServiceSubscriptions[server].clear();
     if (fDebugIsOn)
@@ -813 +770 @@
 bool DataLogger::calculateTotalSizeWritten(DataLoggerStats& statVar, bool isPrinting)
 {
+    if (!isPrinting)
+        fStatsMutex.lock();
 #ifdef HAVE_FITS
     if (isPrinting)
@@ -879 +838 @@
     statVar.sizeWritten -= fBaseSizeNightly;
     statVar.sizeWritten -= fBaseSizeRun;
+
+    if (!isPrinting)
+        fStatsMutex.unlock();
 
     return shouldWarn;
@@ -908 +870 @@
 void DataLogger::ServicesMonitoring()
 {
-        DataLoggerStats statVar;
-        statVar.sizeWritten = 0;
-        statVar.freeSpace = 0;
-        statVar.writingRate = 0;
-
-        struct statvfs vfs;
+       struct statvfs vfs;
         if (!statvfs(fNightlyFilePath.c_str(), &vfs))
-            statVar.freeSpace = vfs.f_bsize*vfs.f_bavail;
+            fStatVar.freeSpace = vfs.f_bsize*vfs.f_bavail;
         else
-            statVar.freeSpace = -1;
-
-        DimDescribedService srvc ("DATA_LOGGER/STATS", "X:3", statVar, "Add description here");
+            fStatVar.freeSpace = -1;
+
+        DimDescribedService srvc ("DATA_LOGGER/STATS", "X:3", fStatVar, "Add description here");
         fPreviousSize = 0;
-        //bool statWarning = false;
-        bool resetDone = false;
         //loop-wait for broadcast
         while (fContinueMonitoring)
         {
-            //check if some run number entries can be deleted
-            while (fRunNumber.size() > 1 && (Time() - fRunNumber.front().time) > boost::posix_time::time_duration(0,0,10,0))
-            {
-                RemoveOldestRunNumber();
-            }
-            //check if daily files should be closed and reopened.
-            //FIXME when Time().h() == 12, the actual time is 2pm
-            if (Time().h() == 12 && !resetDone)
-            {
-                if (fDebugIsOn)
-                    Debug("Change of day detected. Closing daily files and restarting the dataLogger");
-                int cState = GetCurrentState();
-                GoToReadyPlease();
-                //it's noon: no run number should survive
-                //this I'm not too sure about...
-//                while (fRunNumber.size() > 0)
-//                    RemoveOldestRunNumber();
-
-                if (cState >= kSM_NightlyOpen)
-                    StartPlease();
-                if (cState >= kSM_WaitingRun)
-                    NightlyToWaitRunPlease();
-                if (cState >= kSM_Logging)
-                    StartRunPlease();
-
-                resetDone = true;
-            }
-            else
- //           {
- //               if (Time().h() != 12 && resetDone)
-                    resetDone = false;
-//            }
             if (fStatsPeriodDuration == 0.0f)
             {
@@ -966 +889 @@
             sleep(fStatsPeriodDuration);
 
-            //update the fits files sizes
-            /*statWarning =*/ calculateTotalSizeWritten(statVar, false);
-            if (fStatsPeriodDuration == 0.0f)
-                continue;
-            statVar.writingRate = (statVar.sizeWritten - fPreviousSize)/fStatsPeriodDuration;  
-
-            fPreviousSize = statVar.sizeWritten;
-            if (statVar.writingRate != 0) //if data has been written
+
+            fStatsMutex.lock();
+
+            if (fStatVar.writingRate != 0) //if data has been written
             {
                 srvc.updateService();
@@ -986 +905 @@
                 }
             }
+            fStatsMutex.unlock();
         }
 }
@@ -1056 +976 @@
              ("Print information about the internal status of the data logger.");
 
-     //start the monitoring service
-     fContinueMonitoring = true;
-     fMonitoringThread = boost::thread(boost::bind(&DataLogger::ServicesMonitoring, this));
-     fBaseSizeNightly = 0;
-     fBaseSizeRun = 0;
      OpenFileToDim fToDim;
      fToDim.code = 0;
@@ -1108 +1023 @@
 
      fDestructing = false;
+
+     //start the monitoring service
+     fStatVar.sizeWritten = 0;
+     fStatVar.freeSpace = 0;
+     fStatVar.writingRate = 0;
+     fPreviousStatsUpdateTime = Time().Mjd();
+     fPreviousOldRunNumberCheck = Time().Mjd();
+     fContinueMonitoring = true;
+     fMonitoringThread = boost::thread(boost::bind(&DataLogger::ServicesMonitoring, this));
+     fBaseSizeNightly = 0;
+     fBaseSizeRun = 0;
+     fDailyFileDayChangedAlready = true;
      if(fDebugIsOn)
      {
@@ -1197 +1124 @@
     ReportPlease(I, y->second);
 
+    //update the fits files sizes
+    Time cTime = Time();
+    if ((fStatsPeriodDuration != 0) && ((cTime.Mjd() - fPreviousStatsUpdateTime)*24*60*60 > fStatsPeriodDuration))
+    {
+        calculateTotalSizeWritten(fStatVar, false);
+        fStatVar.writingRate = (fStatVar.sizeWritten - fPreviousSize)/((cTime.Mjd() - fPreviousStatsUpdateTime)*24*60*60);
+        fPreviousSize = fStatVar.sizeWritten;
+        fPreviousStatsUpdateTime = cTime.Mjd();
+    }
+    if ((cTime.Mjd() - fPreviousOldRunNumberCheck)*24*60*60 > 10.0)
+    {
+        while (fRunNumber.size() > 1 && (cTime - fRunNumber.back().time) > boost::posix_time::time_duration(0,0,10,0))
+        {
+             RemoveOldestRunNumber();
+        }
+        fPreviousOldRunNumberCheck = cTime.Mjd();
+    }
 }
 // --------------------------------------------------------------------------
@@ -1313 +1257 @@
     if (strstr(I->getName(), fRunNumberInfo) != NULL)
     {//assumes that the run number is an integer
+        //check if some run number entries can be deleted leave one so that two remain after adding the new one
         AddNewRunNumber(I->getLonglong(), Time(I->getTimestamp(), I->getTimestampMillisecs()*1000));
     }
@@ -1327 +1272 @@
 void DataLogger::ReportPlease(DimInfo* I, SubscriptionType& sub)
 {
+
+
     //should we log or report this info ? (i.e. is it a message ?)
     bool isItaReport = ((strstr(I->getName(), "Message") == NULL) && (strstr(I->getName(), "MESSAGE") == NULL));
     if (I->getFormat()[0] == 'C')
         isItaReport = false;
-    
+
     if (!fNightlyReportFile.is_open())
         return;
+
+    //Check whether we should close and reopen daily text files or not
+    //This should work in any case base of the following:
+    // - fDailyFileDayChangedAlready is initialized to true. So if the dataLogger is started around noon, no file will be closed
+    // - fDailyFileDayChangedAlready is set to false if (time != 12), so the file will be closed and reopened only if the logger runs since before noon (which is the required behavior)
+    //This only applies to text files. Fits are closed and reopened based on the last and current service received time.
+    //this was not applicable to text files, because as they gather several services, we have no guarantee that the received time will be greater than the previous one,
+    //which could lead to several close/reopen instead of only one.
+    if (Time().h() == 12 && !fDailyFileDayChangedAlready)
+    {
+        if (fDebugIsOn)
+            Debug("Its Noon! Closing and reopening daily text files");
+
+        fNightlyLogFile.close();
+        fNightlyReportFile.close();
+
+        fFullNightlyLogFileName = CompileFileName(fNightlyFilePath, "", "log");
+        OpenTextFilePlease(fNightlyLogFile, fFullNightlyLogFileName);
+
+        fFullNightlyReportFileName = CompileFileName(fNightlyFilePath, "", "rep");
+        OpenTextFilePlease(fNightlyReportFile, fFullNightlyReportFileName);
+
+        fDailyFileDayChangedAlready = true;
+    }
+    if (Time().h() != 12 && fDailyFileDayChangedAlready)
+        fDailyFileDayChangedAlready = false;
 
     //create the converter for that service
@@ -1466 +1439 @@
     if (isItaReport)
     {
+        //check if the last received event was before noon and if current one is after noon.
+        //if so, close the file so that it gets reopened.
+        if (sub.nightlyFile.IsOpen())
+            if ((sub.lastReceivedEvent != Time::None) && (sub.lastReceivedEvent.h() < 12) && (cTime.h() >= 12))
+            {
+                sub.nightlyFile.Close();
+            }
+        sub.lastReceivedEvent = cTime;
         if (!sub.nightlyFile.IsOpen() || !sub.runFile.IsOpen() || sub.runNumber != sub.runFile.fRunNumber)
             OpenFITSFilesPlease(sub, cRunNumber);
@@ -1520 +1501 @@
 
     DataLoggerStats statVar;
-    /*const bool statWarning =*/ calculateTotalSizeWritten(statVar, false);
+    /*const bool statWarning =*/ calculateTotalSizeWritten(statVar, true);
 
     Message("----------------- STATS ------------------");
@@ -1540 +1521 @@
     str << "There are " << fNumSubAndFitsData.numSubscriptions << " active DIM subscriptions.";
     Message(str);
-
     for (map<const string, map<string, SubscriptionType> >::const_iterator it=fServiceSubscriptions.begin(); it!= fServiceSubscriptions.end();it++)
     {
@@ -1547 +1527 @@
             Message(" -> "+it2->first);
     }
-
     Message("--------------- BLOCK LIST ---------------");
     for (set<string>::const_iterator it=fBlackList.begin(); it != fBlackList.end(); it++)
@@ -1742 +1721 @@
 {
     AddNewRunNumber(evt.GetXtra(), evt.GetTime());
-//    fRunNumber = evt.GetInt();
     return GetCurrentState();
 }
@@ -1867 +1845 @@
         partialName = CompileFileName(fNightlyFilePath, serviceName, "fits");
         fileNameOnly = partialName.substr(partialName.find_last_of('/')+1, partialName.size());
-        AllocateFITSBuffers(sub);
+        if (!sub.fitsBufferAllocated)
+            AllocateFITSBuffers(sub);
         //get the size of the file we're about to open
         if (RememberFileOrigSizePlease(partialName, true))//and remember that the file was opened (i.e. not an update)
@@ -2020 +1999 @@
      sub.nightlyFile.InitDataColumns(GetDescription(sub.server, sub.service), dataFormatsLocal, sub.dimInfo->getData(), size);
      sub.runFile.InitDataColumns(GetDescription(sub.server, sub.service), dataFormatsLocal, sub.dimInfo->getData(), size);
+     sub.fitsBufferAllocated = true;
 }
 // --------------------------------------------------------------------------
@@ -2243 +2223 @@
 {
    if (fDebugIsOn)
-    {
+   {
         Debug("Going to the Ready state...");
-    }    
+   }
    if (GetCurrentState() == kSM_Logging)
        StopRunPlease();