Changeset 10971 for trunk/FACT++/src/dataLogger.cc

Timestamp:

06/10/11 08:52:28 (13 years ago)

Author:

lyard

Message:

Moved monitoring thread to main thread. Added timeout config

File:

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

trunk/FACT++/src/dataLogger.cc

@@ -221 +221 @@
     ///run numbers
     list<RunNumberType> fRunNumber;
+    ///old run numbers time-out delay (in minutes)
+    long fRunNumberTimeout;
     ///previous run number. to check if changed while logging
     int fPreviousRunNumber;
@@ -236 +238 @@
     ///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;
@@ -266 +264 @@
     static const char* fStartStopOpenedFiles;
     static const char* fStartStopNumSubsAndFits;
+    static const char* fSetRunTimeOutDelay;
 private:
     /***************************************************
@@ -315 +314 @@
     void ServicesMonitoring();
     inline void NotifyOpenedFile(const string &name, int type, DimDescribedService* service);
+    ///variables for computing statistics
+    DataLoggerStats fStatVar;
+    ///mutex to make sure that the Stats are not accessed while updating
+//    mutex fStatsMutex;
     ///services notification thread
-    boost::thread fMonitoringThread;
+//    boost::thread fMonitoringThread;
     ///end of the monitoring
-    bool fContinueMonitoring;
+//    bool fContinueMonitoring;
     ///stores the size of each file that is or was open
     map<string, long> fFileSizesMap;
@@ -329 +332 @@
     DimDescribedService* fOpenedRunFiles;
     DimDescribedService* fNumSubAndFits;
+    DimDescribedService* fStatsMonitoring;
     NumSubAndFitsType fNumSubAndFitsData;
     ///Small function for calculating the total size written so far
@@ -351 +355 @@
     int SetOpenedFilesOnOff(const Event& evt);
     int SetNumSubsAndFitsOnOff(const Event& evt);
+    int SetRunTimeoutDelay(const Event& evt);
 
     ///boolean to prevent DIM update while desctructing the dataLogger
@@ -772 +777 @@
 bool DataLogger::calculateTotalSizeWritten(DataLoggerStats& statVar, bool isPrinting)
 {
-    if (!isPrinting)
-        fStatsMutex.lock();
+//mutex
+//    if (!isPrinting)
+//        fStatsMutex.lock();
 #ifdef HAVE_FITS
     if (isPrinting)
@@ -841 +847 @@
     statVar.sizeWritten -= fBaseSizeRun;
 
-    if (!isPrinting)
-        fStatsMutex.unlock();
+//mutex
+//    if (!isPrinting)
+//        fStatsMutex.unlock();
 
     return shouldWarn;
@@ -865 +872 @@
 const char* DataLogger::fStartStopOpenedFiles = "OPENED_FILES_SRVC";
 const char* DataLogger::fStartStopNumSubsAndFits = "NUM_SUBS_SRVC";
+const char* DataLogger::fSetRunTimeOutDelay = "RUN_TIMEOUT";
 
 // --------------------------------------------------------------------------
@@ -871 +879 @@
 //
 //
+/*
 void DataLogger::ServicesMonitoring()
 {
@@ -911 +920 @@
         }
 }
-
+*/
 // --------------------------------------------------------------------------
 //
@@ -1015 +1024 @@
      AddEvent(fStatsPeriod, "F", kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun, kSM_Ready)
              (boost::bind(&DataLogger::SetStatsPeriod, this, _1))
-             ("Interval in which the data-logger statitistics service (STATS) is updated."
+             ("Interval in which the data-logger statistics service (STATS) is updated."
               "Interval[s]:Floating point value in seconds.");
 
@@ -1025 +1034 @@
              (boost::bind(&DataLogger::SetNumSubsAndFitsOnOff, this, _1))
              ("Switch off the service which distributes information about the number of subscriptions and open files.");
+
+     AddEvent(fSetRunTimeOutDelay, "L:1", kSM_Ready, kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun)
+             (boost::bind(&DataLogger::SetRunTimeoutDelay, this, _1))
+             ("Set the timeout delay for old run numbers.");
 
      fDestructing = false;
@@ -1034 +1047 @@
      fPreviousStatsUpdateTime = Time().Mjd();
      fPreviousOldRunNumberCheck = Time().Mjd();
-     fContinueMonitoring = true;
-     fMonitoringThread = boost::thread(boost::bind(&DataLogger::ServicesMonitoring, this));
+     fPreviousSize = 0;
+
+     struct statvfs vfs;
+     if (!statvfs(fNightlyFilePath.c_str(), &vfs))
+         fStatVar.freeSpace = vfs.f_bsize*vfs.f_bavail;
+     else
+         fStatVar.freeSpace = -1;
+
+     fStatsMonitoring = new DimDescribedService(GetName() + "/STATS", "X:3", fStatVar, "Add description here");
+
+//mutex
+//     fContinueMonitoring = true;
+//     fMonitoringThread = boost::thread(boost::bind(&DataLogger::ServicesMonitoring, this));
      fBaseSizeNightly = 0;
      fBaseSizeRun = 0;
      fDailyFileDayChangedAlready = true;
+     fRunNumberTimeout = 1;
      if(fDebugIsOn)
      {
@@ -1061 +1086 @@
     fServiceSubscriptions.clear();
     //exit the monitoring loop
-    fContinueMonitoring = false;
-
-    fMonitoringThread.join();
+//mutex
+//    fContinueMonitoring = false;
+//    fMonitoringThread.join();
     //clear any remaining run number (should remain only one)
      while (fRunNumber.size() > 0)
@@ -1136 +1161 @@
         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))
+        //update the service. No need to check if data has been written, because some must have been, otherwise we would not have hit this piece of code
+        fStatsMonitoring->updateService();
+
+        if(fDebugIsOn)
+        {
+            ostringstream str;
+            str << "Size written: " << fStatVar.sizeWritten/1000 << " kB; writing rate: ";
+            str << fStatVar.writingRate/1000 << " kB/s; free space: ";
+            str << fStatVar.freeSpace/(1000*1000) << " MB";
+            Debug(str);
+        }
+    }
+    if ((cTime.Mjd() - fPreviousOldRunNumberCheck)*24*60 >= fRunNumberTimeout)
+    {
+        while (fRunNumber.size() > 1 && (cTime - fRunNumber.back().time) > boost::posix_time::minutes(fRunNumberTimeout))
         {
              RemoveOldestRunNumber();
@@ -1487 +1523 @@
         str << " <none>";
     Message(str);
+    //timeout value
+    str.str("");
+    str << "Timeout delay for old run numbers: " << fRunNumberTimeout << " minute(s)";
+    Message(str);
 
     //print all the open files. 
@@ -1663 +1703 @@
         ostringstream str;
         str << "Number of subscriptions service mode is now " << fNumSubAndFitsIsOn;
+        Message(str);
+    }
+    return GetCurrentState();
+}
+// --------------------------------------------------------------------------
+//
+//! set the timeout delay for old run numbers
+//! @param evt
+//!        the current event. contains the timeout delay long value
+//! @returns
+//!        the new state. Which, in that case, is the current state
+//!
+int DataLogger::SetRunTimeoutDelay(const Event& evt)
+{
+    const long backupTimeout = fRunNumberTimeout;
+    fRunNumberTimeout = evt.GetXtra();
+
+    if (fRunNumberTimeout == 0)
+    {
+        fRunNumberTimeout = backupTimeout;
+        Error("timeout delays for old run numbers must be greater than 0. Ignored.");
+        return GetCurrentState();
+    }
+
+    if (fRunNumberTimeout == backupTimeout)
+        Warn("Warning: given timeout for old run numbers is same value as previous one");
+    else
+    {
+        ostringstream str;
+        str  << "Timeout delay for old run numbers is now " << fRunNumberTimeout;
         Message(str);
     }
@@ -2314 +2384 @@
     fBlackList.insert("DIS_DNS/");
 
+    //set the black list
     if (conf.Has("block"))
     {
@@ -2321 +2392 @@
     }
 
+    //set the white list
     if (conf.Has("allow"))
     {
@@ -2332 +2404 @@
         const vector<string> vec = conf.Get<vector<string>>("group");
         fGrouping.insert(vec.begin(), vec.end());
+    }
+
+    //set the old run numbers timeout delay
+    if (conf.Has("runtimeout"))
+    {
+        const long timeout = conf.Get<long>("runtimeout");
+        if (timeout != 0)
+            fRunNumberTimeout = timeout;
+        else
+            Error("Time out delay for old run numbers should be greater than 0 minute");
     }
     return true;
@@ -2475 +2557 @@
         ("debug",   po_bool(),      "Debug mode. Print clear text of received service reports to log-stream")
         ("group,g", vars<string>(), "Grouping of services into a single run-Fits")
+        ("runtimeout", var<long>(), "Time out delay for old run numbers")
         ;