Index: /trunk/FACT++/src/dataLogger.cc =================================================================== --- /trunk/FACT++/src/dataLogger.cc (revision 10950) +++ /trunk/FACT++/src/dataLogger.cc (revision 10951) @@ -63,4 +63,5 @@ #include +#include #include @@ -135,36 +136,4 @@ } - ///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,32 +167,9 @@ ///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,12 +178,8 @@ fConv = shared_ptr(); runNumber = 0; - } - ///default constructor -/* SubscriptionType() - { - dimInfo = shared_ptr(); - fConv = shared_ptr(); - runNumber = 0; - }*/ + lastReceivedEvent = Time::None; + fitsBufferAllocated = false; + } + ///default destructor ~SubscriptionType() @@ -293,5 +235,13 @@ ///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,5 +506,5 @@ list[service].server = server; list[service].service = service; - + fNumSubAndFitsData.numSubscriptions++; if (fDebugIsOn) Debug("Added subscription to " + server + "/" + service); @@ -571,6 +521,12 @@ 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,4 +541,5 @@ void DataLogger::RemoveAllServices(const string& server) { + fNumSubAndFitsData.numSubscriptions -= fServiceSubscriptions[server].size(); fServiceSubscriptions[server].clear(); if (fDebugIsOn) @@ -813,4 +770,6 @@ bool DataLogger::calculateTotalSizeWritten(DataLoggerStats& statVar, bool isPrinting) { + if (!isPrinting) + fStatsMutex.lock(); #ifdef HAVE_FITS if (isPrinting) @@ -879,4 +838,7 @@ statVar.sizeWritten -= fBaseSizeNightly; statVar.sizeWritten -= fBaseSizeRun; + + if (!isPrinting) + fStatsMutex.unlock(); return shouldWarn; @@ -908,54 +870,15 @@ 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,12 +889,8 @@ 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,4 +905,5 @@ } } + fStatsMutex.unlock(); } } @@ -1056,9 +976,4 @@ ("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,4 +1023,16 @@ 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,4 +1124,21 @@ 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,4 +1257,5 @@ 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,11 +1272,39 @@ 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,4 +1439,12 @@ 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,5 +1501,5 @@ DataLoggerStats statVar; - /*const bool statWarning =*/ calculateTotalSizeWritten(statVar, false); + /*const bool statWarning =*/ calculateTotalSizeWritten(statVar, true); Message("----------------- STATS ------------------"); @@ -1540,5 +1521,4 @@ str << "There are " << fNumSubAndFitsData.numSubscriptions << " active DIM subscriptions."; Message(str); - for (map >::const_iterator it=fServiceSubscriptions.begin(); it!= fServiceSubscriptions.end();it++) { @@ -1547,5 +1527,4 @@ Message(" -> "+it2->first); } - Message("--------------- BLOCK LIST ---------------"); for (set::const_iterator it=fBlackList.begin(); it != fBlackList.end(); it++) @@ -1742,5 +1721,4 @@ { AddNewRunNumber(evt.GetXtra(), evt.GetTime()); -// fRunNumber = evt.GetInt(); return GetCurrentState(); } @@ -1867,5 +1845,6 @@ 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,4 +1999,5 @@ 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,7 +2223,7 @@ { if (fDebugIsOn) - { + { Debug("Going to the Ready state..."); - } + } if (GetCurrentState() == kSM_Logging) StopRunPlease();