Index: trunk/FACT++/src/Fits.cc =================================================================== --- trunk/FACT++/src/Fits.cc (revision 10439) +++ trunk/FACT++/src/Fits.cc (revision 10442) @@ -18,4 +18,7 @@ #include "Time.h" #include "Converter.h" + +//for file stats +#include using namespace std; @@ -32,4 +35,8 @@ void Fits::AddStandardColumn(Description& desc, std::string dataFormat, void* dataPointer, long unsigned int numDataBytes) { + //check if entry already exist + for (std::vector::iterator it=fStandardColDesc.begin(); it != fStandardColDesc.end(); it++) + if (it->name == desc.name) + return; fStandardColDesc.push_back(desc); fStandardFormats.push_back(dataFormat); @@ -53,4 +60,5 @@ else { + fDataColDesc.clear(); for (unsigned int i=0;imakeThisCurrent(); try { @@ -240,4 +261,5 @@ //This forces the writting of each row to the disk. Otherwise all rows are written when the file is closed. + ///TODO check whether this consumes too much resources or not. If it does, flush every N write operations instead fFile->flush(); } @@ -252,10 +274,11 @@ // if (fTable != NULL) // delete fTable; + std::string name = "TEND"; double doubleValue = fEndMjD; std::string comment = "Time of the last received data"; fTable->addKey(name, doubleValue, comment); - - if (fFile != NULL) + + if (fFile != NULL && fOwner) delete fFile; fFile = NULL; @@ -264,2 +287,19 @@ fCopyBuffer = NULL; } + +// -------------------------------------------------------------------------- +// +//! This closes the currently openned FITS file. +//! it also updates the header to reflect the time of the last logged row +// +int Fits::GetWrittenSize() +{ + if (!IsOpen()) + return 0; + + struct stat st; + if (stat(fFileName.c_str(), &st)) + return 0; + else + return st.st_size; +} Index: trunk/FACT++/src/Fits.h =================================================================== --- trunk/FACT++/src/Fits.h (revision 10439) +++ trunk/FACT++/src/Fits.h (revision 10442) @@ -16,4 +16,6 @@ ///The CCfits object to the FITS file FITS* fFile; + ///Flag indicating whether the FITS object should be managed internally or not. + bool fOwner; ///The CCfits Table Table* fTable; @@ -49,6 +51,7 @@ public: - Fits() : fFile(NULL), - fTable(NULL), + Fits() : fFile(NULL), + fOwner(false), + fTable(NULL), fNumRows(0), fDataPointer(NULL), @@ -56,5 +59,6 @@ fCopyBuffer(NULL), fTotalNumBytes(0), - fEndMjD(0.0) + fEndMjD(0.0), + fFileName("") {} @@ -74,5 +78,5 @@ ///Opens a FITS file - void Open(const std::string& fileName, const std::string& tableName); + void Open(const std::string& fileName, const std::string& tableName, FITS* file); ///Write one line of data. Use the given converter. @@ -81,7 +85,13 @@ ///Close the currently opened file. void Close(); + + ///Get the size currently written on the disk + int GetWrittenSize(); + ///Name of the openned file. For querying stats + std::string fFileName; };//Fits + #endif /*FITS_H_*/ Index: trunk/FACT++/src/dataLogger.cc =================================================================== --- trunk/FACT++/src/dataLogger.cc (revision 10439) +++ trunk/FACT++/src/dataLogger.cc (revision 10442) @@ -54,9 +54,17 @@ #include "Description.h" -#define HAS_FITS +//for getting stat of opened files +#include +//for getting disk free space +#include +//for getting files sizes +#include + +//#define HAS_FITS #include #include +#include #ifdef HAS_FITS @@ -259,8 +267,31 @@ ///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 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,4 +311,89 @@ 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(data), dataSize); + double deltaT = 1; + fPreviousSize = 0; + + long long& targetSize = reinterpret_cast(data)[0]; + long long& targetFreeSpace = reinterpret_cast(data)[1]; + long& targetRate = reinterpret_cast(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::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::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(data), dataSize); + } + } + delete[] data; +} + + // -------------------------------------------------------------------------- // @@ -293,4 +409,7 @@ 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,4 +463,17 @@ 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(fDimBuffer), 256); + } @@ -433,4 +565,9 @@ DataLogger::~DataLogger() { + //exit the monitoring loop + fContinueMonitoring = false; + delete[] fDimBuffer; + delete fOpenedFiles; + fMonitoringThread.join(); //close the files if (fDailyLogFile.is_open()) @@ -444,4 +581,9 @@ //release the services subscriptions fServiceSubscriptions.clear(); +#ifdef HAS_FITS + if (fRunFitsFile != NULL) + delete fRunFitsFile; + fRunFitsFile = NULL; +#endif } // -------------------------------------------------------------------------- @@ -602,4 +744,5 @@ CheckForRunNumber(I); ReportPlease(I, y->second); + } @@ -613,5 +756,4 @@ void DataLogger::CheckForRunNumber(DimInfo* I) { - return; if (strstr(I->getName(), fRunNumberInfo) != NULL) {//assumes that the run number is an integer @@ -822,4 +964,18 @@ // -------------------------------------------------------------------------- // +//! 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(fDimBuffer)[0] = type; + strcpy(&fDimBuffer[sizeof(int)], name.c_str()); + fOpenedFiles->updateService(static_cast(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,9 +989,9 @@ 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,5 +1000,17 @@ 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,5 +1041,16 @@ 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,6 +1058,30 @@ 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,5 +1118,5 @@ dataQualifier << flist[i].second.first; switch (flist[i].first.first->name()[0]) - { + {//TODO handle all the data format cases case 'c': dataQualifier << "S"; @@ -936,4 +1139,5 @@ break; case 'x': + case 'X': dataQualifier << "K"; break; @@ -981,9 +1185,9 @@ 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,4 +1196,20 @@ 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,4 +1235,9 @@ j->second.runFile.Close(); } + if (fRunFitsFile != NULL) + { + delete fRunFitsFile; + fRunFitsFile = NULL; + } #endif return kSM_WaitingRun; @@ -1046,4 +1271,9 @@ j->second.runFile.Close(); } + if (fRunFitsFile != NULL) + { + delete fRunFitsFile; + fRunFitsFile = NULL; + } #endif return kSM_Ready;