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

Timestamp:

04/21/11 12:27:37 (14 years ago)

Author:

lyard

Message:

Added dataLogger services and use only one file for fits runs

File:

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

trunk/FACT++/src/dataLogger.cc

@@ -54 +54 @@
 #include "Description.h"
 
-#define HAS_FITS
+//for getting stat of opened files
+#include <unistd.h>
+//for getting disk free space
+#include <sys/statvfs.h>
+//for getting files sizes
+#include <sys/stat.h>
+
+//#define HAS_FITS
 
 #include <fstream>
 
 #include <boost/bind.hpp>
+#include <boost/thread.hpp>
 
 #ifdef HAS_FITS
@@ -259 +267 @@
         ///Allocate the buffers required for fits
         void AllocateFITSBuffers(SubscriptionType& sub);
+        ///FITS file for runs. only one, hence dealt with in the dataLogger itself
+        FITS* fRunFitsFile;
 #endif
 public: 
         ///checks with fServiceList whether or not the services got updated
         void CheckForServicesUpdate(); 
+
+private:        
+        ///monitoring notification loop
+        void ServicesMonitoring();
+        ///services notification thread
+        boost::thread fMonitoringThread;
+        ///end of the monitoring
+        bool fContinueMonitoring;
+        ///required for accurate monitoring
+        std::map<std::string, long long> fFileSizesMap;
+        std::string fFullDailyLogFileName;
+        std::string fFullDailyReportFileName;
+        std::string fFullRunLogFileName;
+        std::string fFullRunReportFileName;
+        long long fBaseSizeDaily;
+        long long fPreviousSize;
+        long long fBaseSizeRun;
+        ///Service for opened files
+        DimService* fOpenedFiles;
+        char* fDimBuffer;
+        inline void NotifyOpenedFile(std::string name, int type);
         
 }; //DataLogger
@@ -280 +311 @@
 const char* DataLogger::fRunNumberInfo = "RUN_NUMBER";
 
+void DataLogger::ServicesMonitoring()
+{
+                //create the DIM service
+                int dataSize = 2*sizeof(long long) + sizeof(long);
+                char* data = new char[dataSize];
+                memset(data, 0, dataSize);
+                DimService srvc("DATALOGGER/STATS", "x:2;l:1", static_cast<void*>(data), dataSize);
+                double deltaT = 1;
+                fPreviousSize = 0;
+
+                long long& targetSize  = reinterpret_cast<long long*>(data)[0];
+                long long& targetFreeSpace  = reinterpret_cast<long long*>(data)[1];
+                long& targetRate = reinterpret_cast<long*>(data + 2*sizeof(long long))[0];
+                //loop-wait for broadcast
+                while (fContinueMonitoring)
+                {
+                        sleep(deltaT);
+                        //update the fits files sizes
+#ifdef HAS_FITS
+                        SubscriptionsListType::iterator x;
+                        std::map<std::string, SubscriptionType>::iterator y;
+                        bool runFileDone = false;
+                        for (x=fServiceSubscriptions.begin(); x != fServiceSubscriptions.end(); x++)
+                        {
+                                for (y=x->second.begin(); y != x->second.end(); y++)
+                                {
+                                        if (y->second.runFile.IsOpen() && !runFileDone)
+                                        {
+                                                        fFileSizesMap[y->second.runFile.fFileName] = y->second.runFile.GetWrittenSize();
+                                                        runFileDone = true;
+                                        }
+                                        if (y->second.dailyFile.IsOpen())
+                                                fFileSizesMap[y->second.dailyFile.fFileName] = y->second.dailyFile.GetWrittenSize();
+                                }
+                        }
+#endif
+                        struct stat st;
+                        //gather log and report files sizes on disk
+                        if (fDailyLogFile.is_open())
+                        {
+                                stat(fFullDailyLogFileName.c_str(), &st);
+                                fFileSizesMap[fFullDailyLogFileName] = st.st_size;      
+                        }
+                        if (fDailyReportFile.is_open())
+                        {
+                                stat(fFullDailyReportFileName.c_str(), &st);
+                                fFileSizesMap[fFullDailyReportFileName] = st.st_size;   
+                        }
+                        if (fRunLogFile.is_open())
+                        {
+                                stat(fFullRunLogFileName.c_str(), &st);
+                                fFileSizesMap[fFullRunLogFileName] = st.st_size;        
+                        }
+                        if (fRunReportFile.is_open())
+                        {
+                                stat(fFullRunReportFileName.c_str(), &st);
+                                fFileSizesMap[fFullRunReportFileName] = st.st_size;
+                        }       
+                        if (fDailyLogFile.is_open())
+                        {
+                                struct statvfs vfs;
+                                statvfs(fDailyFileName.c_str(), &vfs);
+                                targetFreeSpace = vfs.f_bsize*vfs.f_bavail;
+                        }
+                        //sum up all the file sizes. past and present
+                        targetSize = 0;
+                        for (std::map<std::string, long long>::iterator it=fFileSizesMap.begin(); it != fFileSizesMap.end();  it++)
+                                targetSize += it->second;
+                        targetSize -= fBaseSizeDaily;
+                        targetSize -= fBaseSizeRun;
+                        //FIXME get the actual time elapsed
+                        targetRate = (targetSize - fPreviousSize)/deltaT;  
+                        fPreviousSize = targetSize;
+                        if (targetRate != 0) //if data has been written
+                        {
+std::cout << "fBaseSizeDaily: " << fBaseSizeDaily << " fBaseSizeRun: " << fBaseSizeRun << std::endl;
+//std::cout << "Written Size: " << targetSize << ", free space: " << targetFreeSpace << ", data rate: " << targetRate << " Bytes/s" << std::endl;
+
+                                srvc.updateService(static_cast<void*>(data), dataSize);
+                        }
+                }
+                delete[] data;
+}
+
+
 // --------------------------------------------------------------------------
 //
@@ -293 +409 @@
                 fRunFileName = "/home/lyard/log";
                 fRunNumber = 12345;
+#ifdef HAS_FITS
+                fRunFitsFile = NULL;
+#endif
                 //Give a name to this machine's specific states
                 AddStateName(kSM_DailyOpen,      "DailyFileOpen");
@@ -344 +463 @@
                 fServiceList.SetHandler(this);
                 CheckForServicesUpdate();
+                
+                //start the monitoring service
+                fContinueMonitoring = true;
+                fMonitoringThread = boost::thread(boost::bind(&DataLogger::ServicesMonitoring, this));
+                fBaseSizeDaily = 0;
+                fBaseSizeRun = 0;
+                //start the open files service
+                fDimBuffer = new char[256];
+                memset(fDimBuffer, 0, sizeof(int));
+                fDimBuffer[sizeof(int)] = '\0';
+                //gives the entire buffer size. Otherwise, dim overwrites memory at bizarre locations if smaller size is given at creation time.
+                fOpenedFiles =  new DimService("DATALOGGER/FILENAME", "i:1;C", static_cast<void*>(fDimBuffer), 256);
+                
                 
 }
@@ -433 +565 @@
 DataLogger::~DataLogger()
 {
+        //exit the monitoring loop
+        fContinueMonitoring = false;
+        delete[] fDimBuffer;
+        delete fOpenedFiles;
+        fMonitoringThread.join();
         //close the files
         if (fDailyLogFile.is_open())
@@ -444 +581 @@
         //release the services subscriptions
         fServiceSubscriptions.clear();
+#ifdef HAS_FITS
+        if (fRunFitsFile != NULL)
+                delete fRunFitsFile;
+        fRunFitsFile = NULL;
+#endif
 }
 // --------------------------------------------------------------------------
@@ -602 +744 @@
         CheckForRunNumber(I);
         ReportPlease(I, y->second);
+
 }
 
@@ -613 +756 @@
 void DataLogger::CheckForRunNumber(DimInfo* I)
 {
-        return;
         if (strstr(I->getName(), fRunNumberInfo) != NULL)
         {//assumes that the run number is an integer
@@ -822 +964 @@
 // --------------------------------------------------------------------------
 //
+//! Notifies the DIM service that a particular file was opened
+//! @ param name the base name of the opened file, i.e. without path nor extension. 
+//!     WARNING: use string instead of string& because I pass values that do not convert to string&.
+//!             this is not a problem though because file are not opened so often.
+//! @ param type the type of the opened file. 0 = none open, 1 = log, 2 = text, 4 = fits
+inline void DataLogger::NotifyOpenedFile(std::string name, int type)
+{
+//std::cout << "name: " << name << " size: " << name.size() << std::endl;
+        reinterpret_cast<int*>(fDimBuffer)[0] = type;
+        strcpy(&fDimBuffer[sizeof(int)], name.c_str());
+        fOpenedFiles->updateService(static_cast<void*>(fDimBuffer), name.size() + 1 + sizeof(int));
+}
+// --------------------------------------------------------------------------
+//
 //! Implements the Start transition.
 //! Concatenates the given path for the daily file and the filename itself (based on the day), 
@@ -833 +989 @@
         std::stringstream sTime;
         sTime << time.Y() << "_" << time.M() << "_" << time.D();
-        std::string fullName = fDailyFileName + '/' + sTime.str() + ".log"; 
-        
-        fDailyLogFile.open(fullName.c_str(), std::ios_base::out | std::ios_base::app); //maybe should be "app" instead of "ate" ??
-        fullName = fDailyFileName + '/' + sTime.str() + ".rep";
-        fDailyReportFile.open(fullName.c_str(), std::ios_base::out | std::ios_base::app);
+        fFullDailyLogFileName = fDailyFileName + '/' + sTime.str() + ".log"; 
+        
+        fDailyLogFile.open(fFullDailyLogFileName.c_str(), std::ios_base::out | std::ios_base::app); //maybe should be "app" instead of "ate" ??
+        fFullDailyReportFileName = fDailyFileName + '/' + sTime.str() + ".rep";
+        fDailyReportFile.open(fFullDailyReportFileName.c_str(), std::ios_base::out | std::ios_base::app);
         
         if (!fDailyLogFile.is_open() || !fDailyReportFile.is_open())
@@ -844 +1000 @@
             return kSM_BadDailyConfig;
         }
-
+        //get the size of the newly opened file.
+        struct stat st;
+        stat(fFullDailyLogFileName.c_str(), &st);
+        fBaseSizeDaily = st.st_size;    
+        stat(fFullDailyReportFileName.c_str(), &st);
+        fBaseSizeDaily += st.st_size;
+        fFileSizesMap.clear();
+        fBaseSizeRun = 0;
+        fPreviousSize = 0;
+        //notify that files were opened
+        NotifyOpenedFile(sTime.str(), 3);
+        
+        
         return kSM_DailyOpen;   
 }
@@ -873 +1041 @@
                 std::string partialName = fDailyFileName + '/' + sTime.str() + '_' + serviceName + ".fits";
                 AllocateFITSBuffers(sub);
-                sub.dailyFile.Open(partialName, serviceName);
+                //get the size of the file we're about to open
+                if (fFileSizesMap.find(partialName) == fFileSizesMap.end())
+                {
+                        struct stat st;
+                        if (!stat(partialName.c_str(), &st))
+                                fBaseSizeDaily += st.st_size;
+                        fFileSizesMap[partialName] = 0;
+                }
+                sub.dailyFile.Open(partialName, serviceName, NULL);
+                //notify the opening
+                NotifyOpenedFile(sTime.str() + '_' + serviceName, 4);
+                
         }
         if (!sub.runFile.IsOpen() && (GetCurrentState() == kSM_Logging))
@@ -879 +1058 @@
                 std::stringstream sRun;
                 sRun << fRunNumber;
-                std::string partialName = fRunFileName + '/' + sRun.str() + '_' + serviceName + ".fits";
-                sub.runFile.Open(partialName, serviceName);
+                std::string partialName = fRunFileName + '/' + sRun.str() + ".fits";//'_' + serviceName + ".fits";
+                if (fRunFitsFile == NULL)
+                {
+                        //get the size of the file we're about to open
+                        if (fFileSizesMap.find(partialName) == fFileSizesMap.end())
+                        {
+                                struct stat st;
+                                if (!stat(partialName.c_str(), &st))
+                                        fBaseSizeRun += st.st_size;
+                                else
+                                        fBaseSizeRun = 0;
+                                fFileSizesMap[partialName] = 0; 
+                        }
+                        try
+                        {
+                                fRunFitsFile = new FITS(partialName, RWmode::Write);    
+                        }       
+                        catch (CCfits::FitsError e)
+                        {
+                                std::ostringstream err;
+                                err << "Could not open run file " << partialName;
+                                throw runtime_error(err.str()); 
+                        }
+                        NotifyOpenedFile(sRun.str(), 4);// + '_' + serviceName, 4);
+                }
+                sub.runFile.Open(partialName, serviceName, fRunFitsFile);
         }
 }       
@@ -915 +1118 @@
                 dataQualifier << flist[i].second.first;
                 switch (flist[i].first.first->name()[0])
-                {
+                {//TODO handle all the data format cases
                         case 'c':
                                 dataQualifier <<  "S";
@@ -936 +1139 @@
                         break;
                         case 'x':
+                        case 'X':
                                 dataQualifier << "K";
                         break;
@@ -981 +1185 @@
         std::stringstream sRun;
         sRun << fRunNumber;
-        std::string fullName = fRunFileName + '/' + sRun.str() + ".log";
-        fRunLogFile.open(fullName.c_str(), std::ios_base::out | std::ios_base::app); //maybe should be app instead of ate
-
-        fullName = fRunFileName + '/' + sRun.str() + ".rep";
-        fRunReportFile.open(fullName.c_str(), std::ios_base::out | std::ios_base::app);
+        fFullRunLogFileName = fRunFileName + '/' + sRun.str() + ".log";
+        fRunLogFile.open(fFullRunLogFileName.c_str(), std::ios_base::out | std::ios_base::app); //maybe should be app instead of ate
+
+        fFullRunReportFileName = fRunFileName + '/' + sRun.str() + ".rep";
+        fRunReportFile.open(fFullRunReportFileName.c_str(), std::ios_base::out | std::ios_base::app);
         
         if (!fRunLogFile.is_open() || !fRunReportFile.is_open())
@@ -992 +1196 @@
                 return kSM_BadRunConfig;        
         }
+        //get the size of the newly opened file.
+        struct stat st;
+        fBaseSizeRun = 0;
+        if (fFileSizesMap.find(fFullRunLogFileName) == fFileSizesMap.end())
+        {
+                stat(fFullRunLogFileName.c_str(), &st);
+                fBaseSizeRun += st.st_size;
+                fFileSizesMap[fFullRunLogFileName] = 0;
+        }
+        if (fFileSizesMap.find(fFullRunReportFileName) == fFileSizesMap.end())
+        {
+                stat(fFullRunReportFileName.c_str(), &st);
+                fBaseSizeRun += st.st_size;
+                fFileSizesMap[fFullRunReportFileName] = 0;
+        }
+        NotifyOpenedFile(sRun.str(), 3);
         
         return kSM_Logging;
@@ -1015 +1235 @@
                                         j->second.runFile.Close();      
                 }
+        if (fRunFitsFile != NULL)
+        {
+                delete fRunFitsFile;
+                fRunFitsFile = NULL;    
+        }
 #endif
         return kSM_WaitingRun;
@@ -1046 +1271 @@
                                         j->second.runFile.Close();      
                 }
+        if (fRunFitsFile != NULL)
+        {
+                delete fRunFitsFile;
+                fRunFitsFile = NULL;    
+        }
 #endif
         return kSM_Ready;