Index: /trunk/FACT++/src/Fits.cc =================================================================== --- /trunk/FACT++/src/Fits.cc (revision 10897) +++ /trunk/FACT++/src/Fits.cc (revision 10898) @@ -82,5 +82,5 @@ //! @param fitsCounter a pointer to the integer keeping track of the opened FITS files //! @param out a pointer to the MessageImp that should be used to log errors -//! @param runNumber the runNumber for which this file is opened. -1 means nightly file. +//! @param runNumber the runNumber for which this file is opened. 0 means nightly file. // void Fits::Open(const string& fileName, const string& tableName, FITS* file, int* fitsCounter, MessageImp* out, int runNumber) Index: /trunk/FACT++/src/Fits.h =================================================================== --- /trunk/FACT++/src/Fits.h (revision 10897) +++ /trunk/FACT++/src/Fits.h (revision 10898) @@ -61,5 +61,5 @@ public: ///current run number being logged - int fRunNumber; + uint32_t fRunNumber; Fits() : fFile(NULL), @@ -76,5 +76,5 @@ fNumOpenFitsFiles(NULL), fMess(NULL), - fRunNumber(-1) + fRunNumber(0) {} Index: /trunk/FACT++/src/dataLogger.cc =================================================================== --- /trunk/FACT++/src/dataLogger.cc (revision 10897) +++ /trunk/FACT++/src/dataLogger.cc (revision 10898) @@ -53,6 +53,6 @@ #include "Description.h" -#include "DimServiceInfoList.h" - +//#include "DimServiceInfoList.h" +#include "DimNetwork.h" //for getting stat of opened files #include @@ -109,5 +109,5 @@ string reportName; ///the actual run number - int runNumber; + uint32_t runNumber; ///the time at which the run number was received Time time; @@ -124,5 +124,5 @@ runFitsFile = NULL; #endif - runNumber = -1; + runNumber = 0; //give it a meaningless, 0 time to distinguish with actual, valid times time = Time(0,0); @@ -221,4 +221,8 @@ ///the actual dimInfo pointer DimStampedInfo* dimInfo; + ///the server + string server; + ///the service + string service; ///the converter for outputting the data according to the format Converter* fConv; @@ -226,5 +230,5 @@ int* numCopies; ///the current run number used by this subscription - int runNumber; + uint32_t runNumber; ///copy operator void operator = (const SubscriptionType& other) @@ -235,4 +239,6 @@ #endif dimInfo = other.dimInfo; + server = other.server; + service = other.service; numCopies = other.numCopies; fConv = other.fConv; @@ -248,4 +254,6 @@ #endif dimInfo = other.dimInfo; + server = other.server; + service = other.service; numCopies = other.numCopies; fConv = other.fConv; @@ -258,5 +266,5 @@ dimInfo = info; fConv = NULL; - runNumber = -1; + runNumber = 0; numCopies = new int(1); } @@ -266,5 +274,5 @@ dimInfo = NULL; fConv = NULL; - runNumber = -1; + runNumber = 0; numCopies = new int(1); } @@ -295,5 +303,5 @@ }; -class DataLogger : public StateMachineDim, DimInfoHandler +class DataLogger : public StateMachineDim, DimServiceInfoList { public: @@ -334,5 +342,5 @@ double fMjD; ///for obtaining the name of the existing services - ServiceList fServiceList; +// ServiceList fServiceList; typedef map > SubscriptionsListType; ///All the services to which we have subscribed to, sorted by server name. @@ -382,10 +390,8 @@ ///DEPREC - configuration of the run number int ConfigureRunNumber(const Event& evt); - ///logging method for the messages -// int LogMessagePlease(const Event& evt); ///print the current state of the dataLogger int PrintStatePlease(const Event& evt); ///checks whether or not the current info being treated is a run number - void CheckForRunNumber(DimInfo* I); + void CheckForRunNumber(DimInfo* I); /// start transition int StartPlease(); @@ -462,9 +468,7 @@ int OpenRunFile(RunNumberType& run); ///add a new run number - void AddNewRunNumber(int newRun, Time time); + void AddNewRunNumber(int64_t newRun, Time time); ///removes the oldest run number, and close the relevant files. void RemoveOldestRunNumber(); - ///checks with fServiceList whether or not the services got updated - bool CheckForServicesUpdate(); ///retrieves the size of a file off_t GetFileSize(string&); @@ -497,6 +501,91 @@ ///Checks if the input osftream is in error state, and if so close it. void CheckForOfstreamError(ofstream& out); + /*************************************************** + * INHERITED FROM DIMSERVICEINFOLIST + ***************************************************/ + ///Add a new service subscription + void AddService(const string&, const string&, const string&, bool); + ///Remove a given service subscription + void RemoveService(const string&, const string&, bool); + ///Remove all the services associated with a given server + void RemoveAllServices(const string&); }; //DataLogger +// -------------------------------------------------------------------------- +// +//! Add a new service subscription +//! @param server the server for which the subscription should be created +//! @param service the service for which the subscription should be created +//! @param isCmd whether this is a Dim Command or not. Commands are not logged +//TODO maybe commands should indeed be logged ? +// +void DataLogger::AddService(const string& server, const string& service, const string&, bool isCmd) +{ + //check the given subscription against black and white lists + if (!ShouldSubscribe(server, service)) + return; + //dataLogger does not subscribe to commands + if (isCmd) + return; + //do we already have an entry for that server ? + SubscriptionsListType::iterator cSubs = fServiceSubscriptions.find(server); + if (cSubs != fServiceSubscriptions.end()) + {//server already here. check its services + if (cSubs->second.find(service) == cSubs->second.end()) + {//service not found. Add it + cSubs->second[service].dimInfo = SubscribeToPlease(server, service); + cSubs->second[service].server = server; + cSubs->second[service].service = service; + } + else + { + MessageImp::Error("Service " + server + "/" + service + " is already in the dataLogger's list. ignoring its update."); + return; + } + } + else + {//server not yet in our list. add it + fServiceSubscriptions[server] = map(); + map& liste = fServiceSubscriptions[server]; + liste[service].dimInfo = SubscribeToPlease(server, service); + liste[service].server = server; + liste[service].service = service; + } + if (fDebugIsOn) + { + Debug("Added subscription to " + server + "/" + service); + } +} +// -------------------------------------------------------------------------- +// +//! Remove a given service subscription +//! @param server the server for which the subscription should be removed +//! @param service the service that should be removed +//! @param isCmd whether or not this is a command +// +void DataLogger::RemoveService(const string& server, const string& service, bool isCmd) +{ + if (isCmd) + return; + + fServiceSubscriptions[server].erase(service); + if (fDebugIsOn) + { + Debug("Removed subscription to " + server + "/" + service); + } +} +// -------------------------------------------------------------------------- +// +//! Remove all the services associated with a given server +//! @param server the server for which all the services should be removed +// +void DataLogger::RemoveAllServices(const string& server) +{ + fServiceSubscriptions[server].clear(); + if (fDebugIsOn) + { + Debug("Removed all subscriptions to " + server + "/"); + } +} // -------------------------------------------------------------------------- // @@ -627,4 +716,5 @@ fileNameWithPath = fRunFilePath + '/' + fileName; } + // -------------------------------------------------------------------------- // @@ -637,5 +727,5 @@ { ostringstream groupName; - if (runNumber != -1) + if (runNumber != 0) { groupName << fRunFilePath << '/' << runNumber << ".fits"; @@ -750,5 +840,5 @@ return st.st_size; - if (errno != 0) + if (errno != 0 && errno != 2)//ignoring error #2: no such file or directory is not an error for new files { ostringstream str; @@ -1002,18 +1092,8 @@ "|Path[string]:Absolute or relative path name where the run files should be stored."); - AddEvent(fConfigRunNumber, "I", kSM_Ready, kSM_NightlyOpen, kSM_WaitingRun, kSM_BadRunConfig, kSM_Logging) + AddEvent(fConfigRunNumber, "X", kSM_Ready, kSM_NightlyOpen, kSM_WaitingRun, kSM_BadRunConfig, kSM_Logging) (boost::bind(&DataLogger::ConfigureRunNumber, this, _1)) ("Configure the run number. Cannot be done in logging state"); - //Provide a logging command - //I get the feeling that I should be going through the EventImp - //instead of DimCommand directly, mainly because the commandHandler - //is already done in StateMachineImp.cc - //Thus I'll simply add a configuration, which I will treat as the logging command -// AddEvent(fConfigLog, "C", kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun, kSM_BadRunConfig) - // (boost::bind(&DataLogger::LogMessagePlease, this, _1)) - // ("Log a single message to the log-files." -// "|Message[string]:Message to be logged."); - //Provide a print command ostringstream str; @@ -1023,7 +1103,4 @@ (boost::bind(&DataLogger::PrintStatePlease, this, _1)) ("Print information about the internal status of the data logger."); - - fServiceList.SetHandler(this); - CheckForServicesUpdate(); //start the monitoring service @@ -1087,91 +1164,5 @@ } } -// -------------------------------------------------------------------------- -// -//! Checks for changes in the existing services. -//! Any new service will be added to the service list, while the ones which disappeared are removed. -// -//FIXME The service must be udpated so that I get the first notification. This should not be -bool DataLogger::CheckForServicesUpdate() -{ - bool serviceUpdated = false; - //get the current server list - const vector serverList = fServiceList.GetServerList(); - //first let's remove the servers that may have disapeared - //can't treat the erase on maps the same way as for vectors. Do it the safe way instead - vector toBeDeleted; - for (SubscriptionsListType::iterator cListe = fServiceSubscriptions.begin(); cListe != fServiceSubscriptions.end(); cListe++) - { - vector::const_iterator givenServers; - for (givenServers=serverList.begin(); givenServers!= serverList.end(); givenServers++) - if (cListe->first == *givenServers) - break; - if (givenServers == serverList.end())//server vanished. Remove it - { - toBeDeleted.push_back(cListe->first); - serviceUpdated = true; - } - } - for (vector::const_iterator it = toBeDeleted.begin(); it != toBeDeleted.end(); it++) - fServiceSubscriptions.erase(*it); - //now crawl through the list of servers, and see if there was some updates - for (vector::const_iterator i=serverList.begin(); i!=serverList.end();i++) - { - //skip the two de-fact excluded services - //Dim crashes if the publisher subscribes to its own service. This sounds weird, I agree. - //I agree: hard coded values are bad, but in this case these two entries should always be here otherwise Dim crashes. - if ((i->find("DIS_DNS") != string::npos) || - (i->find("DATA_LOGGER") != string::npos)) - continue; - //find the current server in our subscription list - SubscriptionsListType::iterator cSubs = fServiceSubscriptions.find(*i); - //get the service list of the current server - vector cServicesList = fServiceList.GetServiceList(*i); - if (cSubs != fServiceSubscriptions.end())//if the current server already is in our subscriptions - { //then check and update our list of subscriptions - //first, remove the services that may have disapeared. - map::iterator serverSubs; - vector::const_iterator givenSubs; - toBeDeleted.clear(); - for (serverSubs=cSubs->second.begin(); serverSubs != cSubs->second.end(); serverSubs++) - { - for (givenSubs = cServicesList.begin(); givenSubs != cServicesList.end(); givenSubs++) - if (serverSubs->first == *givenSubs) - break; - if (givenSubs == cServicesList.end()) - { - toBeDeleted.push_back(serverSubs->first); - serviceUpdated = true; - } - } - for (vector::const_iterator it = toBeDeleted.begin(); it != toBeDeleted.end(); it++) - cSubs->second.erase(*it); - //now check for new services - for (givenSubs = cServicesList.begin(); givenSubs != cServicesList.end(); givenSubs++) - { - if (!ShouldSubscribe(*i, *givenSubs)) - continue; - if (cSubs->second.find(*givenSubs) == cSubs->second.end()) - {//service not found. Add it - cSubs->second[*givenSubs].dimInfo = SubscribeToPlease(*i, *givenSubs); - serviceUpdated = true; - } - } - } - else //server not found in our list. Create its entry - { - fServiceSubscriptions[*i] = map(); - map& liste = fServiceSubscriptions[*i]; - for (vector::const_iterator j = cServicesList.begin(); j!= cServicesList.end(); j++) - { - if (!ShouldSubscribe(*i, *j)) - continue; - liste[*j].dimInfo = SubscribeToPlease(*i, *j); - serviceUpdated = true; - } - } - } - return serviceUpdated; -} + // -------------------------------------------------------------------------- // @@ -1220,31 +1211,13 @@ DimInfo* I = getInfo(); - SubscriptionsListType::iterator x; - map::iterator y; + if (I==NULL) - { - if (CheckForServicesUpdate()) - { - //services were updated. Notify - fNumSubAndFitsData.numSubscriptions = 0; - for (x=fServiceSubscriptions.begin(); x != fServiceSubscriptions.end(); x++) - fNumSubAndFitsData.numSubscriptions += x->second.size(); - if (fNumSubAndFitsIsOn) - { - if (fDebugIsOn) - { - ostringstream str; - str << "Updating number of subscriptions service: Num Subs=" << fNumSubAndFitsData.numSubscriptions << " Num open FITS=" << fNumSubAndFitsData.numOpenFits; - Debug(str); - } - fNumSubAndFits->updateService(); - } - } return; - } //check if the service pointer corresponds to something that we subscribed to //this is a fix for a bug that provides bad Infos when a server starts bool found = false; - for (x=fServiceSubscriptions.begin(); x != fServiceSubscriptions.end(); x++) + SubscriptionsListType::iterator x; + map::iterator y; + for (x=fServiceSubscriptions.begin(); x != fServiceSubscriptions.end(); x++) {//find current service is subscriptions for (y=x->second.begin(); y!=x->second.end();y++) @@ -1258,5 +1231,8 @@ } if (!found) + { + DimServiceInfoList::infoHandler(); return; + } if (I->getSize() <= 0) return; @@ -1336,6 +1312,12 @@ //! @param time the time at which the new run number was issued // -void DataLogger::AddNewRunNumber(int newRun, Time time) -{ +void DataLogger::AddNewRunNumber(int64_t newRun, Time time) +{ + + if (newRun > 0xffffffff) + { + Error("New run number too large, out of range. Ignoring."); + return; + } if (fDebugIsOn) { @@ -1346,6 +1328,11 @@ //Add new run number to run number list fRunNumber.push_back(RunNumberType()); - fRunNumber.back().runNumber = newRun; + fRunNumber.back().runNumber = uint32_t(newRun); fRunNumber.back().time = time; + + ostringstream str; + str << "The new run number is: " << fRunNumber.back().runNumber; + Message(str); + if (GetCurrentState() != kSM_Logging) return; @@ -1364,8 +1351,5 @@ if (strstr(I->getName(), fRunNumberInfo) != NULL) {//assumes that the run number is an integer - AddNewRunNumber(I->getInt(), Time(I->getTimestamp(), I->getTimestampMillisecs()*1000)); - ostringstream str; - str << "New run number is " << fRunNumber.back().runNumber; - Message(str); + AddNewRunNumber(I->getLonglong(), Time(I->getTimestamp(), I->getTimestampMillisecs()*1000)); } } @@ -1523,102 +1507,4 @@ } -// -------------------------------------------------------------------------- -// -//! write messages to logs. -/* -//! @param evt -//! the current event to log -//! @returns -//! the new state. Currently, always the current state -//! -//! @deprecated -//! I guess that this function should not be any longer -// -//Otherwise re-write it properly with the MessageImp class -int DataLogger::LogMessagePlease(const Event& evt) -{ - if (!fNightlyLogFile.is_open()) - return GetCurrentState(); - Warn("LogMessagePlease has not been checked nor updated since a long while. Undefined behavior to be expected"); - ostringstream header; - const Time& cTime = evt.GetTime(); - header << evt.GetName() << " " << cTime.Y() << " " << cTime.M() << " " << cTime.D() << " "; - header << cTime.h() << " " << cTime.m() << " " << cTime.s() << " "; - header << cTime.ms() << " "; - - const Converter conv(Out(), evt.GetFormat()); - if (!conv) - { - Error("Couldn't properly parse the format... ignored."); - return GetCurrentState(); - } - - string text; - try - { - text = conv.GetString(evt.GetData(), evt.GetSize()); - } - catch (const runtime_error &e) - { - Out() << kRed << e.what() << endl; - Error("Couldn't properly parse the data... ignored."); - return GetCurrentState(); - } - - if (text.empty()) - return GetCurrentState(); - - //replace bizarre characters by white space - replace(text.begin(), text.end(), '\n', '\\'); - replace_if(text.begin(), text.end(), ptr_fun(&iscntrl), ' '); - if (fDebugIsOn) - { - ostringstream str; - str << "Logging: \"" << header << text << "\""; - Debug(str.str()); - } - - if (fNightlyLogFile.is_open()) - { - fNightlyLogFile << header; - if (!fNightlyLogFile.good()) - { - Error("An error occured while writing to the run log file. Closing it."); - if (fNightlyLogFile.is_open()) - fNightlyLogFile.close(); - } - } - if (fRunLogFile.is_open()) - { - fRunLogFile << header; - if (!fRunLogFile.good()) - { - Error("An error occured while writing to the run log file. Closing it."); - if (fRunLogFile.is_open()) - fRunLogFile.close(); - } - } - if (fNightlyLogFile.is_open()) - { - fNightlyLogFile << text; - if (!fNightlyLogFile.good()) - { - Error("An error occured while writing to the run log file. Closing it."); - if (fNightlyLogFile.is_open()) - fNightlyLogFile.close(); - } - } - if (fRunLogFile.is_open()) - { - fRunLogFile << text; - if (!fRunLogFile.good()) - { - Error("An error occured while writing to the run log file. Closing it."); - if (fRunLogFile.is_open()) - fRunLogFile.close(); - } - } - return GetCurrentState(); -}*/ // -------------------------------------------------------------------------- // @@ -1878,9 +1764,6 @@ int DataLogger::ConfigureRunNumber(const Event& evt) { - AddNewRunNumber(evt.GetInt(), evt.GetTime()); + AddNewRunNumber(evt.GetXtra(), evt.GetTime()); // fRunNumber = evt.GetInt(); - ostringstream str; - str << "The new run number is: " << fRunNumber.back().runNumber; - Message(str); return GetCurrentState(); } @@ -2009,5 +1892,5 @@ fOpenedNightlyFits[fileNameOnly].push_back(serviceName); - sub.nightlyFile.Open(partialName, serviceName, NULL, &fNumSubAndFitsData.numOpenFits, this, -1);//Out()); + sub.nightlyFile.Open(partialName, serviceName, NULL, &fNumSubAndFitsData.numOpenFits, this, 0);//Out()); //notify the opening @@ -2025,5 +1908,5 @@ } //do the actual file open - if (!sub.runFile.IsOpen() && (GetCurrentState() == kSM_Logging) && sub.runNumber != -1) + if (!sub.runFile.IsOpen() && (GetCurrentState() == kSM_Logging) && sub.runNumber != 0) {//buffer for the run file have already been allocated when doing the Nightly file string fileNameOnly; @@ -2156,6 +2039,6 @@ dataFormatsLocal.push_back(dataQualifier.str()); } - sub.nightlyFile.InitDataColumns(fServiceList.GetDescriptions(sub.dimInfo->getName()), dataFormatsLocal, sub.dimInfo->getData(), size); - sub.runFile.InitDataColumns(fServiceList.GetDescriptions(sub.dimInfo->getName()), dataFormatsLocal, sub.dimInfo->getData(), size); + 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); } // -------------------------------------------------------------------------- @@ -2210,5 +2093,5 @@ //! @param filesToGroup a map of filenames mapping to table names to be grouped (i.e. a //! single file can contain several tables to group -//! @param runNumber the run number that should be used for grouping. -1 means nightly group +//! @param runNumber the run number that should be used for grouping. 0 means nightly group // void DataLogger::CreateFitsGrouping(map > & filesToGroup, int runNumber) @@ -2218,5 +2101,5 @@ ostringstream str; str << "Creating fits group for "; - if (runNumber != -1) + if (runNumber != 0) str << "run files"; else @@ -2397,5 +2280,5 @@ } #ifdef HAVE_FITS - CreateFitsGrouping(fOpenedNightlyFits, -1); + CreateFitsGrouping(fOpenedNightlyFits, 0); #endif return kSM_Ready;