Changeset 14233 for trunk

Timestamp:

06/26/12 12:37:25 (12 years ago)

Author:

lyard

Message:

replaced DimServiceInfoListImp with EventImps. Warning, currently the logger only subscribes to services that exist when it is launched

Location:

trunk/FACT++/src

Files:

• Fits.cc (modified) (3 diffs)
• Fits.h (modified) (4 diffs)
• datalogger.cc (modified) (45 diffs)

trunk/FACT++/src/Fits.cc

@@ -60 +60 @@
 //! @param out Message object to use for propagating messages
 //      
-void Fits::InitDataColumns(const vector<Description> &desc, const vector<string>& dataFormat, void* dataPointer, MessageImp* out)
+void Fits::InitDataColumns(const vector<Description> &desc, const vector<string>& dataFormat, const void* dataPointer, MessageImp* out)
 {
     fDataFormats = dataFormat;
-    fDataPointer = dataPointer;
+//FIXME I removed the static assignment of the data pointer because it changes from event to event now. Remove the arg of this function
+//    fDataPointer = dataPointer;
 
     if ((desc.size() == 0) && (dataFormat.size() == 0))
@@ -314 +315 @@
 //! @param conv the converter corresponding to the service being logged
 //
-bool Fits::Write(const Converter &conv)
+bool Fits::Write(const Converter &conv, const void* data)
 {
     //first copy the standard variables to the copy buffer
@@ -328 +329 @@
     {
         //now take care of the DIM data. The Converter is here for that purpose
-        conv.ToFits(fCopyBuffer.data()+shift, fDataPointer, fCopyBuffer.size()-shift);
+        conv.ToFits(fCopyBuffer.data()+shift, data, fCopyBuffer.size()-shift);
     }
     catch (const runtime_error &e)

trunk/FACT++/src/Fits.h

@@ -39 +39 @@
 
     ///the pointer to the contiguous memory location where the data is stored (i.e. the dim data pointer)
-    void* fDataPointer;
+//    void* fDataPointer;
     ///the copy buffer. Required to put the standard and data variable in contguous memory
     vector<char> fCopyBuffer;
@@ -61 +61 @@
 
     Fits() : fFile(NULL),
-        fDataPointer(NULL),
         fEndMjD(0.0),
         fNumOpenFitsFiles(NULL),
@@ -80 +79 @@
 
     ///Adds columns specific to the service being logged.
-    void InitDataColumns(const vector<Description> &desc, const vector<string>& dataFormat, void* dataPointer, MessageImp* out);
+    void InitDataColumns(const vector<Description> &desc, const vector<string>& dataFormat, const void* dataPointer, MessageImp* out);
 
     ///Opens a FITS file
@@ -86 +85 @@
 
     ///Write one line of data. Use the given converter.
-    bool Write(const Converter &conv);
+    bool Write(const Converter &conv, const void* data);
 
     ///Close the currently opened file.

trunk/FACT++/src/datalogger.cc

@@ -84 +84 @@
 #endif
 
+#include "DimState.h"
 
 //Dim structures
@@ -115 +116 @@
     }
 };
+
+EventImp nullEventImp;
 ///Dim subscription type. Stores all the relevant info to handle a Dim subscription
 struct SubscriptionType
@@ -137 +140 @@
     /// that it is the first which is deleted -- and consequently none of
     /// the other members can still be in use in an infoHandler)
-    shared_ptr<DimStampedInfo> dimInfo;
+    //DIM_REPLACE
+    //shared_ptr<DimStampedInfo> dimInfo;
+    unsigned int index;
+
     ///Dim info constructor
-    SubscriptionType(DimStampedInfo* info=NULL)
+    //DIM_REPLACE
+//    SubscriptionType(DimStampedInfo* info=NULL)
+    SubscriptionType()
     {
         fConv = shared_ptr<Converter>();
@@ -145 +153 @@
         lastReceivedEvent = Time::None;
         fitsBufferAllocated = false;
-
         // Should be the last instantiated to make sure that all other
         // variables which might be used are already initialized
-        dimInfo = shared_ptr<DimStampedInfo>(info);
+        //DIM_REPLACE
+        //dimInfo = shared_ptr<DimStampedInfo>(info);
+        index = 0;
     }
     ///default destructor
@@ -156 +165 @@
 };
 
-class DataLogger : public StateMachineDim, DimServiceInfoListImp
+class DataLogger : public StateMachineDim
+//DIM_REPLACE
+//, DimServiceInfoListImp
 {
 public:
@@ -216 +227 @@
      ***************************************************/
     //overloading of DIM's infoHandler function
-    void infoHandler(); 
+    int infoCallback(const EventImp& evt, unsigned int infoIndex);
 
     /***************************************************
@@ -222 +233 @@
      ***************************************************/
     ///Reporting method for the services info received
-    void Report(DimInfo* I, SubscriptionType& sub);
+    void Report(const EventImp& evt, SubscriptionType& sub);
 
     ///Configuration of the nightly file path
@@ -229 +240 @@
     int PrintState(const Event& evt);
     ///checks whether or not the current info being treated is a run number
-    void CheckForRunNumber(DimInfo* I);
+    void CheckForRunNumber(const EventImp& evt, unsigned int index);
     /// start transition
     int Start();
@@ -244 +255 @@
 #ifdef HAVE_FITS
     ///Open fits files
-    void OpenFITSFiles(SubscriptionType& sub);
+    void OpenFITSFiles(SubscriptionType& sub, const EventImp& evt);
     ///Write data to FITS files
-    void WriteToFITS(SubscriptionType& sub);
+    void WriteToFITS(SubscriptionType& sub, const void* data);
     ///Allocate the buffers required for fits
-    void AllocateFITSBuffers(SubscriptionType& sub);
+    void AllocateFITSBuffers(SubscriptionType& sub, const EventImp& evt);
 #endif//has_fits
 
@@ -314 +325 @@
     bool ShouldSubscribe(const string& server, const string& service);
     ///Subscribe to a given server and service
-    DimStampedInfo* SubscribeTo(const string& server, const string& service);
+//    EventImp& SubscribeTo(const string& server, const string& service);
     ///Open a text file and checks for ofstream status
     bool OpenTextFile(ofstream& stream, const string& name);
@@ -331 +342 @@
     * INHERITED FROM DimServiceInfoList
     ***************************************************/
+    ///Add a new server subscription
+    void AddServer(const string& server);
     ///Add a new service subscription
-    void AddService(const string&, const string&, const string&, bool);
+    void AddService(const Service& svc);
     ///Remove a given service subscription
+    //FIXME unused
     void RemoveService(const string, const string, bool);
     ///Remove all the services associated with a given server
+    //FIXME unused
     void RemoveAllServices(const string&);
     ///pointer to the dim's subscription that should distribute the run numbers.
-    DimInfo* fRunNumberService;
+    //DIM_REPLACE
+    //DimInfo* fRunNumberService;
+    unsigned int fRunNumberService;
     /***************************************************
      * Overwritten from MessageImp
@@ -350 +367 @@
     int fCurrentDay;
     Time lastFlush;
+
+    DimDnsServiceList fDimList;
+    vector<DimDescriptions*> fServerDescriptionsList;
+
+    //counter for keeping tracker of services
+    unsigned int servicesCounter;
 public:
     int Write(const Time &time, const std::string &txt, int qos=kMessage);
@@ -412 +435 @@
 }
 
+
+void DataLogger::AddServer(const string& server)
+{
+    Info("Got request to add server " + server );
+    if (server != "DIS_DNS")
+    {
+        DimDescriptions* d = new DimDescriptions(server);
+        d->Subscribe(*this);
+        fServerDescriptionsList.push_back(d);
+    }
+
+}
+
 // --------------------------------------------------------------------------
 //
@@ -419 +455 @@
 //! @param isCmd whether this is a Dim Command or not. Commands are not logged
 //
-void DataLogger::AddService(const string& server, const string& service, const string&, bool isCmd)
-{
-    //dataLogger does not subscribe to commands
+void DataLogger::AddService(const Service& svc)
+{
+    const string& server = svc.server;
+    const string& service = svc.service;
+    const bool isCmd = svc.iscmd;
+
+    Info("Got request to add service: "+server+"/"+service);
+   //dataLogger does not subscribe to commands
     if (isCmd)
         return;
 
-    Info("Got request to add service: "+server+"/"+service);
 
     //check the given subscription against black and white lists
@@ -438 +478 @@
         return;
     }
-
-    list[service].dimInfo.reset(SubscribeTo(server, service));
+    //DIM_REPLACE
+//    list[service].dimInfo.reset(SubscribeTo(server, service));
+    if (fDebugIsOn)
+        Debug("Subscribing to service "+server+"/"+service);
+    Subscribe(server + "/" + service)
+        (bind(&DataLogger::infoCallback, this, placeholders::_1, servicesCounter));
     list[service].server  = server;
     list[service].service = service;
+    list[service].index = servicesCounter;
     fNumSubAndFitsData.numSubscriptions++;
     //check if this is the run numbers service
     if ((server == "FAD_CONTROL") && (service == "START_RUN"))
-        fRunNumberService = list[service].dimInfo.get();
+        fRunNumberService = servicesCounter;
+    servicesCounter++;
     Info("Added subscription to " + server + "/" + service);
 }
@@ -457 +503 @@
 void DataLogger::RemoveService(string server, string service, bool isCmd)
 {
+
     Info("Got request to remove service: "+server+"/"+service);
     if (fDestructing)//this function is called by the super class, after the destructor has deleted its own subscriptions
         return;
+//FIXME unused
+    return;
 
     if (isCmd)
@@ -482 +531 @@
 
     if ((server == "FAD_CONTROL") && (service == "START_RUN"))
-        fRunNumberService = NULL;
+        fRunNumberService = 0;
 
     Info("Removed subscription to " + server + "/" + service);
@@ -499 +548 @@
         return;
     }
-
+//FIXME unused
+    return;
     fNumSubAndFitsData.numSubscriptions -= fServiceSubscriptions[server].size();
 
@@ -506 +556 @@
 
     if (server == "FAD_CONTROL")
-        fRunNumberService = NULL;
+        fRunNumberService = 0;
 
     if (fDebugIsOn)
@@ -585 +635 @@
 //! @param service the service name
 //
-DimStampedInfo* DataLogger::SubscribeTo(const string& server, const string& service)
-{
-    if (fDebugIsOn)
-        Debug("Subscribing to service "+server+"/"+service);
-
-    return new DimStampedInfo((server + "/" + service).c_str(), (void*)NULL, 0, this);
-}
+/*EventImp& DataLogger::SubscribeTo(const string& server, const string& service)
+{
+
+    //DIM_REPLACE
+    //return new DimStampedInfo((server + "/" + service).c_str(), (void*)NULL, 0, this);
+    EventImp& newSubscription = Subscribe(server + "/" + service);
+    newSubscription.bind(&infoHandler, this, placeholders::_1);
+    return newSubscription;
+}*/
 // --------------------------------------------------------------------------
 //
@@ -703 +755 @@
 {
     shouldBackLog = true;
+
+    servicesCounter=1;
+
     //initialize member data
     fFilePath = ".";
+
+    fDimList.Subscribe(*this);
+    fDimList.SetCallbackServerAdd(bind(&DataLogger::AddServer, this, placeholders::_1));
+    fDimList.SetCallbackServiceAdd(bind(&DataLogger::AddService, this, placeholders::_1));
 
     //calculate time "centered" around noon instead of midnight
@@ -809 +868 @@
      NotifyOpenedFile("", 0, fOpenedRunFiles);
 
-     fRunNumberService = NULL;
+     fRunNumberService = 0;
+
      fShouldAutoStart = false;
      fAutoStarted = false;
+
+
      if(fDebugIsOn)
      {
@@ -855 +917 @@
     }
 
+    for (auto it=fServerDescriptionsList.begin(); it!= fServerDescriptionsList.end(); it++)
+        delete *it;
+
     if (fDebugIsOn)
         Debug("DataLogger desctruction ends");    
@@ -880 +945 @@
 //! Inherited from DimInfo. Handles all the Infos to which we subscribed, and log them
 //
-void DataLogger::infoHandler()
-{
+int DataLogger::infoCallback(const EventImp& evt, unsigned int subIndex)
+{
+//    if (fDebugIsOn)
+//    {
+//        ostringstream str;
+//        str << "Got infoCallback called with service index= " << subIndex;
+//        Debug(str.str());
+//    }
     if ((GetCurrentState() == kSM_Ready) &&  (!fAutoStarted) && fShouldAutoStart)
     {
@@ -892 +963 @@
             fAutoStarted = true;
     }
-    DimInfo* I = getInfo();
-
-    if (I==NULL)
-        return;
+
 
     //check if the service pointer corresponds to something that we subscribed to
@@ -904 +972 @@
     for (x=fServiceSubscriptions.begin(); x != fServiceSubscriptions.end(); x++)
     {//find current service is subscriptions
+     //Edit: this should be useless now... remove it sometimes ?
         for (y=x->second.begin(); y!=x->second.end();y++)
-            if ((y->second.dimInfo).get() == I)
+            if (y->second.index == subIndex)
             {
                 found = true;    
@@ -914 +983 @@
     }
     if (!found)
-        return;
-
-    if (I->getSize() <= 0 || I->getData()==NULL)
-    {
-        return;
-    }
-    if (strlen(I->getFormat()) == 0)
-    {
-        ostringstream str;
-        str << "Format of " << I->getName() << " is empty (ptr=" << I->getData() << ", size=" << I->getSize() << ")... ignoring it.";
-        Warn(str);
-        return;
-    }
-    // Make sure that getTimestampMillisecs is NEVER called before
-    // getTimestamp is properly called
-    // check that the message has been updated by something, i.e. must be different from its initial value
-    if (I->getTimestamp() == 0)
-    {
-        return;
-    }
+        return GetCurrentState();
+
+    if (evt.GetSize() == 0)
+        return GetCurrentState();
+    if (evt.GetFormat() == "")
+        return GetCurrentState();
+
 //    cout.precision(20);
 //    cout << "Orig timestamp: " << Time(I->getTimestamp(), I->getTimestampMillisecs()*1000).Mjd() << endl;
@@ -941 +997 @@
     //        subscribe to this service anyway and hence we have the pointer
     //        (no need to check for the name)
-    CheckForRunNumber(I);
-
-    Report(I, y->second);
+    CheckForRunNumber(evt, subIndex);
+
+    Report(evt, y->second);
 
     //remove old run numbers
     TrimOldRunNumbers();
+
+    return GetCurrentState();
 }
 
@@ -1002 +1060 @@
 //!        the current DimInfo
 //
-void DataLogger::CheckForRunNumber(DimInfo* I)
-{
-    if (I != fRunNumberService)
+void DataLogger::CheckForRunNumber(const EventImp& evt, unsigned int index)
+{
+    if (index != fRunNumberService)
         return;
-
-    AddNewRunNumber(I->getLonglong(), Time(I->getTimestamp(), I->getTimestampMillisecs()*1000));
+//    int64_t newRun = reinterpret_cast<const uint64_t*>(evt.GetData())[0];
+    AddNewRunNumber(evt.GetXtra(), evt.GetTime());
 }
 
@@ -1018 +1076 @@
 //!        The dataLogger's subscription corresponding to this DimInfo
 //
-void DataLogger::Report(DimInfo* I, SubscriptionType& sub)
-{
-    const string fmt(I->getFormat());
+void DataLogger::Report(const EventImp& evt, SubscriptionType& sub)
+{
+    const string fmt(evt.GetFormat());
 
     const bool isItaReport = fmt!="C";
@@ -1027 +1085 @@
         return;
 
-    if (fDebugIsOn && string(I->getName())!="DATA_LOGGER/MESSAGE")
+    if (fDebugIsOn && string(evt.GetName())!="DATA_LOGGER/MESSAGE")
     {
         ostringstream str;
-        str << "Logging " << I->getName() << " [" << I->getFormat() << "] (" << I->getSize() << ")";
+        str << "Logging " << evt.GetName() << " [" << evt.GetFormat() << "] (" << evt.GetSize() << ")";
         Debug(str);
     }
@@ -1105 +1163 @@
     if (!sub.fConv)
     {
-        sub.fConv = shared_ptr<Converter>(new Converter(Out(), I->getFormat()));
+        sub.fConv = shared_ptr<Converter>(new Converter(Out(), evt.GetFormat()));
         if (!sub.fConv->valid())
         {
             ostringstream str;
-            str << "Couldn't properly parse the format... service " << sub.dimInfo->getName() << " ignored.";
+            str << "Couldn't properly parse the format... service " << evt.GetName() << " ignored.";
             Error(str);
             return;    
@@ -1116 +1174 @@
     //construct the header
     ostringstream header;
-    const Time cTime(I->getTimestamp(), I->getTimestampMillisecs()*1000);
-    fQuality = I->getQuality();
+    const Time cTime(evt.GetTime());
+    fQuality = evt.GetQoS();
     //I had strange surprises with the quality from Dim before. Double check that the value is indeed valid.
     if (fQuality != kMessage &&
@@ -1132 +1190 @@
     {
         //write text header
-        header << I->getName() << " " << fQuality << " ";
+        header << evt.GetName() << " " << fQuality << " ";
         header << cTime.Y() << " " << cTime.M() << " " << cTime.D() << " ";
         header << cTime.h() << " " << cTime.m() << " " << cTime.s() << " ";
-        header << cTime.ms() << " " << I->getTimestamp() << " ";
+        header << cTime.ms() << " " << evt.GetTime() << " ";
 
         string text;
         try
         {
-            text = sub.fConv->GetString(I->getData(), I->getSize());
+            text = sub.fConv->GetString(evt.GetData(), evt.GetSize());
         }
         catch (const runtime_error &e)
         {
             ostringstream str;
-            str << "Parsing service " << sub.dimInfo->getName();
+            str << "Parsing service " << evt.GetName();
             str << " failed: " << e.what() << " removing the subscription for now.";
             Error(str);
@@ -1157 +1215 @@
         {
             ostringstream str;
-            str << "Service " << sub.dimInfo->getName() << " sent an empty string";
+            str << "Service " << evt.GetName() << " sent an empty string";
             Info(str);
             return;
@@ -1178 +1236 @@
         if (!sub.nightlyFile.IsOpen())
             if (GetCurrentState() != kSM_Ready)
-                OpenFITSFiles(sub);
-        WriteToFITS(sub);
+                OpenFITSFiles(sub, evt);
+        WriteToFITS(sub, evt.GetData());
 #endif
     }
@@ -1187 +1245 @@
         try
         {
-           strings = sub.fConv->ToStrings(I->getData());
+           strings = sub.fConv->ToStrings(evt.GetData());
         }
         catch (const runtime_error &e)
         {
             ostringstream str;
-            str << "Parsing service " << sub.dimInfo->getName();
+            str << "Parsing service " << evt.GetName();
             str << " failed: " << e.what() << " removing the subscription for now.";
             Error(str);
@@ -1203 +1261 @@
         {
             ostringstream err;
-            err << "There was more than one string message in service " << I->getName() << " going to fatal error state";
+            err << "There was more than one string message in service " << evt.GetName() << " going to fatal error state";
             Error(err.str());
         }
         ostringstream msg;
-        msg << I->getName() << ": " << strings[0];
+        msg << evt.GetName() << ": " << strings[0];
 
         if (fNightlyLogFile.is_open())
@@ -1219 +1277 @@
         if (!sub.nightlyFile.IsOpen())
             if (GetCurrentState() != kSM_Ready)
-                OpenFITSFiles(sub);
-        WriteToFITS(sub);
+                OpenFITSFiles(sub, evt);
+        WriteToFITS(sub, evt.GetData());
     }
 
@@ -1525 +1583 @@
 //! @param sub
 //!     the current DimInfo subscription being examined
-void DataLogger::OpenFITSFiles(SubscriptionType& sub)
-{
-    string serviceName(sub.dimInfo->getName());
+void DataLogger::OpenFITSFiles(SubscriptionType& sub, const EventImp& evt)
+{
+    string serviceName(sub.server + "_" + sub.service);//evt.GetName());
 
     for (unsigned int i=0;i<serviceName.size(); i++)
@@ -1544 +1602 @@
         const string fileNameOnly = partialName.substr(partialName.find_last_of('/')+1, partialName.size());
         if (!sub.fitsBufferAllocated)
-            AllocateFITSBuffers(sub);
+            AllocateFITSBuffers(sub, evt);
         //get the size of the file we're about to open
         if (fFilesStats.FileOpened(partialName))
@@ -1572 +1630 @@
 //! @param sub the subscription of interest.
 //
-void DataLogger::AllocateFITSBuffers(SubscriptionType& sub)
+void DataLogger::AllocateFITSBuffers(SubscriptionType& sub, const EventImp& evt)
 {
     //Init the time columns of the file
@@ -1595 +1653 @@
     str << "Initializing data columns for service " << sub.server << "/" << sub.service;
     Info(str);
-    sub.nightlyFile.InitDataColumns(GetDescription(sub.server, sub.service), dataFormatsLocal, sub.dimInfo->getData(), this);
+    sub.nightlyFile.InitDataColumns(Description::SplitDescription(evt.GetDescription()), dataFormatsLocal, evt.GetData(), this);
 
     sub.fitsBufferAllocated = true;
@@ -1603 +1661 @@
 //! write a dimInfo data to its corresponding FITS files
 //
-void DataLogger::WriteToFITS(SubscriptionType& sub)
+//FIXME: DO I REALLY NEED THE EVENT IMP HERE ???
+void DataLogger::WriteToFITS(SubscriptionType& sub, const void* data)
 {
         //nightly File status (open or not) already checked
         if (sub.nightlyFile.IsOpen())
         {
-            if (!sub.nightlyFile.Write(*sub.fConv.get()))
+            if (!sub.nightlyFile.Write(*sub.fConv.get(), data))
             {
                 RemoveService(sub.server, sub.service, false);
@@ -1976 +2035 @@
 int RunShell(Configuration &conf)
 {
-    return Main::execute<T, DataLogger>(conf, true);
+    return Main::execute<T, DataLogger>(conf);//, true);
 }