Changeset 11019 for trunk/FACT++/src/scheduler.cc

Timestamp:

06/14/11 18:04:59 (13 years ago)

Author:

tbretz

Message:

Many structural updates; catch mysqlpp exceptions outside Scheudle(); added error state; added reset command to get out of the error state

File:

• trunk/FACT++/src/scheduler.cc (modified) (27 diffs)

trunk/FACT++/src/scheduler.cc

@@ -9 +9 @@
 #include <boost/thread.hpp>
 #include <boost/regex.hpp>
-#include <boost/asio/deadline_timer.hpp>
-#include <boost/date_time/posix_time/posix_time.hpp>
 
 #include <mysql++/mysql++.h>
@@ -28 +26 @@
 #include "tools.h"
 
-namespace ba    = boost::asio;
-namespace bs    = boost::system;
-
-using ba::deadline_timer;
-using ba::ip::tcp;
-
 using namespace std;
 using namespace boost::gregorian;
@@ -41 +33 @@
 // * do not use --database, but sth like --database-scheduler
 //   to be independent of the configuration database
-// * check if the following include is really needed
-//   #include <boost/date_time/posix_time/posix_time.hpp>
-// * string database not as global variable
-//   pass reference to conf and the string as a member of the class
-//   (see chat of 4.5.2011)
 // * move definition of config parameters to AutoScheduler class
 //   + read in from config
@@ -98 +85 @@
         float ra;
         float dec;
-        boost::posix_time::ptime starttime;
-        boost::posix_time::ptime stoptime;
-        boost::posix_time::time_duration duration_db;
+        ptime start;
+        ptime stop;
+        time_duration duration_db;
         string sourcename;
         int sourcekey;
@@ -107 +94 @@
     struct FixedObs
     {
-        int obskey_fixed;
-        int sourcekey_fixed;
-        string sourcename_fixed;
-        int obsmode_fixed;
-        int obstype_fixed;
-        int telsetup_fixed;
-        float ra_fixed;
-        float dec_fixed;
-        boost::posix_time::ptime obsfixedstart;
-        boost::posix_time::ptime obsfixedstop;
+        int obskey;
+        int sourcekey;
+        string sourcename;
+        int obsmode;
+        int obstype;
+        int telsetup;
+        float ra;
+        float dec;
+        ptime start;
+        ptime stop;
     };
 
@@ -137 +124 @@
         float ra_std;
         float dec_std;
-        boost::posix_time::ptime obsstdstart;
-        boost::posix_time::ptime obsstdstop;
+        ptime obsstdstart;
+        ptime obsstdstop;
     };
 
@@ -148 +135 @@
         int obstype_obs;
         int telsetup_obs;
-        boost::posix_time::ptime obsstart;
-        boost::posix_time::ptime obsstop;
+        ptime obsstart;
+        ptime obsstop;
     };
 
@@ -161 +148 @@
         int obstype_run;
         int telsetup_run;
-        boost::posix_time::ptime runstart;
-        boost::posix_time::ptime runstop;
+        ptime runstart;
+        ptime runstop;
     };
 
@@ -175 +162 @@
         bool error = false;
 
-        boost::posix_time::time_duration runtimec(0, 3, 0);
-        boost::posix_time::time_duration runtimep(0, 3, 0);
-        boost::posix_time::time_duration mintime(1, 0, 0);
-        boost::posix_time::time_duration repostime(0, 5, 0);
-        //boost::posix_time::ptime startsched=microsec_clock::local_time();
-        boost::posix_time::ptime startsched(microsec_clock::local_time());
-        boost::posix_time::ptime stopsched=startsched+years(1);
-        cout << "Scheduling for the period from " << startsched << " to " << stopsched << endl;
+        time_duration runtimec(0, 3, 0);
+        time_duration runtimep(0, 3, 0);
+        time_duration mintime(1, 0, 0);
+        time_duration repostime(0, 5, 0);
+        //ptime startsched=microsec_clock::local_time();
+        ptime startsched(microsec_clock::local_time());
+        ptime stopsched=startsched+years(1);
+
+        ostringstream str;
+        str << "Scheduling for the period from " << startsched << " to " << stopsched;
+        T::Message(str);
 
         static const boost::regex expr("(([[:word:].-]+)(:(.+))?@)?([[:word:].-]+)(:([[:digit:]]+))?(/([[:word:].-]+))");
@@ -194 +184 @@
         if (!boost::regex_match(fDatabase, what, expr, boost::match_extra))
         {
-            cout << "Couldn't parse '" << fDatabase << "'." << endl;
-            throw;
+            ostringstream msg;
+            msg << "Regex to parse database '" << fDatabase << "' empty.";
+            T::Error(msg);
+            return T::kSM_Error;
         }
 
         if (what.size()!=10)
         {
-            cout << "Error parsing '" << fDatabase << "'." << endl;
-            throw;
+            ostringstream msg;
+            msg << "Parsing database name failed: '" << fDatabase << "'";
+            T::Error(msg);
+            return T::kSM_Error;
         }
 
@@ -216 +210 @@
         T::Message(dbnamemsg);
 
-        cout << "Connecting to '";
+        str.str("");
+        str << "Connecting to '";
         if (!user.empty())
-            cout << user << "@";
-        cout << server;
+            str << user << "@";
+        str << server;
         if (port)
-            cout << ":" << port;
+            str << ":" << port;
         if (!db.empty())
-            cout << "/" << db;
-        cout << "'" << endl;
+            str << "/" << db;
+        str << "'";
+        T::Info(str);
 
         mysqlpp::Connection conn(db.c_str(), server.c_str(), user.c_str(), passwd.c_str(), port);
+        /* throws exceptions
         if (!conn.connected())
         {
-            cout << "MySQL connection error: " << conn.error() << endl;
-            throw;
-        }
+            ostringstream msg;
+            msg << "MySQL connection error: " << conn.error();
+            T::Error(msg);
+            return T::kSM_Error;
+        }*/
 
         // get observation parameters from DB
         // maybe order by priority?
-        mysqlpp::Query query = conn.query("SELECT fObservationKEY, fStartTime, fStopTime, fDuration, fSourceName, fSourceKEY, fSplitFlag, fFluxWeight, fSlope, fFlux, fRightAscension, fDeclination, fObservationModeKEY, fObservationTypeKEY , fTelescopeSetupKEY FROM ObservationParameters LEFT JOIN Source USING(fSourceKEY) ORDER BY fStartTime");
-
-        mysqlpp::StoreQueryResult res = query.store();
+        const mysqlpp::StoreQueryResult res =
+            conn.query("SELECT fObservationKEY, fStartTime, fStopTime, fDuration, fSourceName, fSourceKEY, fSplitFlag, fFluxWeight, fSlope, fFlux, fRightAscension, fDeclination, fObservationModeKEY, fObservationTypeKEY , fTelescopeSetupKEY FROM ObservationParameters LEFT JOIN Source USING(fSourceKEY) ORDER BY fStartTime").store();
+        // FIXME: Maybe we have to check for a successfull
+        //        query but an empty result
+        /* thorws exceptions?
         if (!res)
         {
-            cout << "MySQL query failed: " << query.error() << endl;
-            throw;
-        }
-
-        cout << "Found " << res.num_rows() << " Observation Parameter sets." << endl;
+            ostringstream msg;
+            msg << "MySQL query failed: " << query.error();
+            T::Error(msg);
+            return T::kSM_Error;
+        }*/
+
+        str.str("");
+        str << "Found " << res.num_rows() << " Observation Parameter sets.";
+        T::Debug(str);
 
         ObservationParameters olist[res.num_rows()];
-        std::vector<FixedObs> obsfixedlist;
-        std::vector<StdObs> obsstdlist;
-        std::vector<ScheduledObs> obslist;
-        std::vector<ScheduledRun> runlist;
+        vector<FixedObs> obsfixedlist;
+        vector<StdObs> obsstdlist;
+        vector<ScheduledObs> obslist;
+        vector<ScheduledRun> runlist;
 
         // loop over observation parameters from DB
@@ -258 +263 @@
         int counter3=0;
         cout << "Obs: <obskey> <sourcename>(<sourcekey>, <fluxweight>) from <starttime> to <stoptime>" << endl;
-        for (vector<mysqlpp::Row>::iterator v=res.begin(); v<res.end(); v++)
+        for (vector<mysqlpp::Row>::const_iterator v=res.begin(); v<res.end(); v++)
         {
             cout << "  Obs: " << (*v)[0].c_str() << " " << (*v)[4].c_str() << "(" << (*v)[5].c_str() << flush;
@@ -287 +292 @@
 
             //boost::posix_time::time_duration mintime(0,conf.Get<int>("mintime"), 0);
-            t1 >> Time::sql >> olist[counter].starttime;
-            t2 >> Time::sql >> olist[counter].stoptime;
+            t1 >> Time::sql >> olist[counter].start;
+            t2 >> Time::sql >> olist[counter].stop;
             t3 >> olist[counter].duration_db;
-            boost::posix_time::time_period period(olist[counter].starttime, olist[counter].stoptime);
+            const time_period period(olist[counter].start, olist[counter].stop);
+
             olist[counter].sourcename=(*v)[4].c_str();
             olist[counter].sourcekey=(*v)[5];
@@ -318 +324 @@
 
             // time_duration cannot be used, as only up to 99 hours are handeled
-            boost::posix_time::time_duration duration = period.length();
+            const time_duration duration = period.length();
 
             /*
@@ -354 +360 @@
 
             // fixed observations
-            if (!(olist[counter].stoptime.is_not_a_date_time()
-                  && olist[counter].starttime.is_not_a_date_time())
+            if (!(olist[counter].stop.is_not_a_date_time()
+                  && olist[counter].start.is_not_a_date_time())
                 && olist[counter].fluxweight==0
                )
             {
                 obsfixedlist.resize(counter2+1);
-                obsfixedlist[counter2].obsfixedstart=olist[counter].starttime;
-                obsfixedlist[counter2].obsfixedstop=olist[counter].stoptime;
-                obsfixedlist[counter2].sourcename_fixed=olist[counter].sourcename;
-                obsfixedlist[counter2].obskey_fixed=olist[counter].obskey;
-                obsfixedlist[counter2].obstype_fixed=olist[counter].obstype;
-                obsfixedlist[counter2].obsmode_fixed=olist[counter].obsmode;
-                obsfixedlist[counter2].telsetup_fixed=olist[counter].telsetup;
-                obsfixedlist[counter2].sourcekey_fixed=olist[counter].sourcekey;
-                obsfixedlist[counter2].ra_fixed=olist[counter].ra;
-                obsfixedlist[counter2].dec_fixed=olist[counter].dec;
+                obsfixedlist[counter2].start=olist[counter].start;
+                obsfixedlist[counter2].stop=olist[counter].stop;
+                obsfixedlist[counter2].sourcename=olist[counter].sourcename;
+                obsfixedlist[counter2].obskey=olist[counter].obskey;
+                obsfixedlist[counter2].obstype=olist[counter].obstype;
+                obsfixedlist[counter2].obsmode=olist[counter].obsmode;
+                obsfixedlist[counter2].telsetup=olist[counter].telsetup;
+                obsfixedlist[counter2].sourcekey=olist[counter].sourcekey;
+                obsfixedlist[counter2].ra=olist[counter].ra;
+                obsfixedlist[counter2].dec=olist[counter].dec;
                 counter2++;
             }
 
             // std obs
-            if (olist[counter].stoptime.is_not_a_date_time()
-                  && olist[counter].starttime.is_not_a_date_time()
+            if (olist[counter].stop.is_not_a_date_time()
+                  && olist[counter].start.is_not_a_date_time()
                 && olist[counter].fluxweight>0
                )
@@ -416 +422 @@
         counter2=0;
         int skipcounter=0;
-        boost::posix_time::ptime finalobsfixedstart;
-        boost::posix_time::ptime finalobsfixedstop;
-        boost::posix_time::time_duration delta1;
-        boost::posix_time::time_duration delta2;
-        boost::posix_time::time_duration delta0(0,0,0);
-        struct vector<FixedObs>::iterator vobs;
-        struct vector<FixedObs>::iterator vobs5;
+        ptime finalobsfixedstart;
+        ptime finalobsfixedstop;
+        time_duration delta0(0,0,0);
+
         cout << "Fixed Observations: " << endl;
-        for (vobs=obsfixedlist.begin(); vobs!=obsfixedlist.end(); vobs++)
-        {
-            if (obsfixedlist[counter2].obsfixedstart < startsched
-                || obsfixedlist[counter2].obsfixedstop > stopsched)
+        for (struct vector<FixedObs>::const_iterator vobs=obsfixedlist.begin(); vobs!=obsfixedlist.end(); vobs++)
+        {
+            if (obsfixedlist[counter2].start < startsched
+                || obsfixedlist[counter2].stop > stopsched)
             {
                 ostringstream skipfixedobsmsg;
-                skipfixedobsmsg  << "Skip 1 fixed observation (obskey " << obsfixedlist[counter2].obskey_fixed << ") as it is out of scheduling time range.";
+                skipfixedobsmsg << "Skip 1 fixed observation (obskey ";
+                skipfixedobsmsg << obsfixedlist[counter2].obskey;
+                skipfixedobsmsg << ") as it is out of scheduling time range.";
                 T::Message(skipfixedobsmsg);
+
                 counter2++;
                 skipcounter++;
@@ -437 +443 @@
             }
             counter3=0;
-            delta1=delta0;
-            delta2=delta0;
-            finalobsfixedstart=obsfixedlist[counter2].obsfixedstart;
-            finalobsfixedstop=obsfixedlist[counter2].obsfixedstop;
-
-            for (vobs5=obsfixedlist.begin(); vobs5!=obsfixedlist.end(); vobs5++)
+
+            time_duration delta1=delta0;
+            time_duration delta2=delta0;
+
+            finalobsfixedstart=obsfixedlist[counter2].start;
+            finalobsfixedstop=obsfixedlist[counter2].stop;
+
+            for (struct vector<FixedObs>::const_iterator vobs5=obsfixedlist.begin(); vobs5!=obsfixedlist.end(); vobs5++)
             {
-                if ((*vobs5).obsfixedstart < obsfixedlist[counter2].obsfixedstop
-                    && obsfixedlist[counter2].obsfixedstop <= (*vobs5).obsfixedstop
-                    && obsfixedlist[counter2].obsfixedstart <= (*vobs5).obsfixedstart
+                if (vobs5->start < obsfixedlist[counter2].stop
+                    && obsfixedlist[counter2].stop <= vobs5->stop
+                    && obsfixedlist[counter2].start <= vobs5->start
                     && counter2!=counter3)
                 {
-                    delta1=(obsfixedlist[counter2].obsfixedstop-(*vobs5).obsfixedstart)/2;
-                    finalobsfixedstop=obsfixedlist[counter2].obsfixedstop-delta1;
+                    delta1=(obsfixedlist[counter2].stop-vobs5->start)/2;
+                    finalobsfixedstop=obsfixedlist[counter2].stop-delta1;
+
                     ostringstream warndelta1;
-                    warndelta1 << "Overlap between two fixed observations (" << obsfixedlist[counter2].obskey_fixed << " " << (*vobs5).obskey_fixed << "). The stoptime of " << obsfixedlist[counter2].obskey_fixed << " has been changed.";
+                    warndelta1 << "Overlap between two fixed observations (";
+                    warndelta1 << obsfixedlist[counter2].obskey << " ";
+                    warndelta1 << vobs5->obskey << "). The stoptime of ";
+                    warndelta1 << obsfixedlist[counter2].obskey << " has been changed.";
                     T::Warn(warndelta1);
                 }
-                if ((*vobs5).obsfixedstart <= obsfixedlist[counter2].obsfixedstart
-                    && obsfixedlist[counter2].obsfixedstart < (*vobs5).obsfixedstop
-                    && obsfixedlist[counter2].obsfixedstop >= (*vobs5).obsfixedstop
+                if (vobs5->start <= obsfixedlist[counter2].start
+                    && obsfixedlist[counter2].start < vobs5->stop
+                    && obsfixedlist[counter2].stop >= vobs5->stop
                     && counter2!=counter3)
                 {
-                    delta2=((*vobs5).obsfixedstop-obsfixedlist[counter2].obsfixedstart)/2;
-                    finalobsfixedstart=obsfixedlist[counter2].obsfixedstart+delta2;
+                    delta2=(vobs5->stop-obsfixedlist[counter2].start)/2;
+                    finalobsfixedstart=obsfixedlist[counter2].start+delta2;
+
                     ostringstream warndelta2;
-                    warndelta2 << "Overlap between two fixed observations (" << obsfixedlist[counter2].obskey_fixed << " " << (*vobs5).obskey_fixed << "). The starttime of " << obsfixedlist[counter2].obskey_fixed << " has been changed.";
+                    warndelta2 << "Overlap between two fixed observations (";
+                    warndelta2 << obsfixedlist[counter2].obskey << " ";
+                    warndelta2 << vobs5->obskey << "). The starttime of ";
+                    warndelta2 << obsfixedlist[counter2].obskey << " has been changed.";
+
                     T::Warn(warndelta2);
                 }
@@ -469 +486 @@
             }
 
-            int num=counter2-skipcounter;
+            const int num=counter2-skipcounter;
             obslist.resize(num+1);
             obslist[num].obsstart=finalobsfixedstart;
             obslist[num].obsstop=finalobsfixedstop;
-            obslist[num].sourcename_obs=obsfixedlist[counter2].sourcename_fixed;
-            obslist[num].obsmode_obs=obsfixedlist[counter2].obsmode_fixed;
-            obslist[num].obstype_obs=obsfixedlist[counter2].obstype_fixed;
-            obslist[num].telsetup_obs=obsfixedlist[counter2].telsetup_fixed;
-            obslist[num].sourcekey_obs=obsfixedlist[counter2].sourcekey_fixed;
+            obslist[num].sourcename_obs=obsfixedlist[counter2].sourcename;
+            obslist[num].obsmode_obs=obsfixedlist[counter2].obsmode;
+            obslist[num].obstype_obs=obsfixedlist[counter2].obstype;
+            obslist[num].telsetup_obs=obsfixedlist[counter2].telsetup;
+            obslist[num].sourcekey_obs=obsfixedlist[counter2].sourcekey;
             counter2++;
-            cout << "  " << (*vobs).sourcename_fixed <<  " " << (*vobs).obsfixedstart << flush;
-            cout << " - " << (*vobs).obsfixedstop << endl;
+
+            cout << "  " << vobs->sourcename <<  " " << vobs->start;
+            cout << " - " << vobs->stop << endl;
         }
         ostringstream obsmsg;
@@ -512 +530 @@
         // the observation times shall be calculated
         // and the observations added to the ScheduledObs list
-        struct vector<StdObs>::iterator vobs2;
         cout << "Standard Observations: " << endl;
-        for (vobs2=obsstdlist.begin(); vobs2!=obsstdlist.end(); vobs2++)
-        {
-            cout << "  " << (*vobs2).sourcename_std << endl;
+        for (struct vector<StdObs>::const_iterator vobs2=obsstdlist.begin(); vobs2!=obsstdlist.end(); vobs2++)
+        {
+            cout << "  " << vobs2->sourcename_std << endl;
         }
 
@@ -523 +540 @@
         // might be merged with next loop
         counter2=0;
-        struct vector<ScheduledObs>::iterator vobs3;
-        for (vobs3=obslist.begin(); vobs3!=obslist.end(); vobs3++)
+        for (struct vector<ScheduledObs>::const_iterator vobs3=obslist.begin(); vobs3!=obslist.end(); vobs3++)
         {
             runlist.resize(counter2+1);
@@ -539 +555 @@
 
         //delete old scheduled runs from the DB
-        mysqlpp::Query query0 = conn.query("DELETE FROM ScheduledRun");
-
-        mysqlpp::SimpleResult res0 = query0.execute();
+        const mysqlpp::SimpleResult res0 =
+            conn.query("DELETE FROM ScheduledRun").execute();
+        // FIXME: Maybe we have to check for a successfull
+        //        query but an empty result
+        /* throws exceptions
         if (!res0)
         {
-            cout << "MySQL query failed: " << query0.error() << endl;
-            throw;
-        }
+            ostringstream msg;
+            msg << "MySQL query failed: " << query0.error();
+            T::Error(msg);
+            return T::kSM_Error;
+        }*/
 
         // in this loop the ScheduledRuns are inserted to the DB
@@ -553 +573 @@
         counter3=0;
         int insertcount=0;
-        boost::posix_time::ptime finalstarttime;
-        boost::posix_time::ptime finalstoptime;
-        struct vector<ScheduledRun>::iterator vobs4;
-        for (vobs4=runlist.begin(); vobs4!=runlist.end(); vobs4++)
+        ptime finalstarttime;
+        ptime finalstoptime;
+        for (struct vector<ScheduledRun>::const_iterator vobs4=runlist.begin(); vobs4!=runlist.end(); vobs4++)
         {
             for (int i=2; i<5; i++)
@@ -588 +607 @@
                 //cout << "executing query: " << q1.str() << endl;
 
-                mysqlpp::Query query1 = conn.query(q1.str());
-
-                mysqlpp::SimpleResult res1 = query1.execute();
+                const mysqlpp::SimpleResult res1 = conn.query(q1.str()).execute();
+                // FIXME: Maybe we have to check for a successfull
+                //        query but an empty result
+                /* throws exceptions
                 if (!res1)
                 {
-                    cout << "MySQL query failed: " << query1.error() << endl;
-                    throw;
-                }
+                    ostringstream msg;
+                    msg << "MySQL query failed: " << query1.error();
+                    T::Error(str);
+                    return T::kSM_Error;
+                }*/
                 insertcount++;
             }
@@ -608 +630 @@
         fSessionId = -1;
 
-        //bool error = false;
         return error ? T::kSM_Error : T::kSM_Ready;
     }
@@ -632 +653 @@
     AutoScheduler(ostream &out=cout) : T(out, "SCHEDULER"), fNextIsPreview(true), fSessionId(-1), fDBName("")
     {
-        AddStateName(kSM_Scheduling,  "Scheduling");
-        //AddStateName(kSM_Comitting,   "Comitting");
-
-        AddEvent(kSM_Scheduling, "SCHEDULE", T::kSM_Ready);
+        AddStateName(kSM_Scheduling, "Scheduling", "Scheduling in progress.");
+
+        AddEvent(kSM_Scheduling, "SCHEDULE", "C", T::kSM_Ready)
+            ("FIXME FIXME FIXME (explanation for the command)"
+             "|database[string]:FIXME FIXME FIMXE (meaning and format)");
+
+        AddEvent(T::kSM_Ready, "RESET", T::kSM_Error)
+            ("Reset command to get out of the error state");
+
         //AddEvent(kSM_Comitting,  "COMMIT",   T::kSM_Ready);
 
@@ -646 +672 @@
         {
         case kSM_Scheduling:
-            return Schedule();
+            try
+            {
+                return Schedule();
+            }
+            catch (const mysqlpp::Exception &e)
+            {
+                T::Error(string("MySQL: ")+e.what());
+                return T::kSM_Error;
+            }
+
+            // This does an autmatic reset (FOR TESTING ONLY)
+        case T::kSM_Error:
+            return T::kSM_Ready;
 
         //case kSM_Comitting:
@@ -811 +849 @@
         vm = conf.Parse(argc, argv);
     }
-    catch (std::exception &e)
-    {
 #if BOOST_VERSION > 104000
-        po::multiple_occurrences *MO = dynamic_cast<po::multiple_occurrences*>(&e);
-        if (MO)
-            cout << "Error: " << e.what() << " of '" << MO->get_option_name() << "' option." << endl;
-        else
+    catch (po::multiple_occurrences &e)
+    {
+        cerr << "Program options invalid due to: " << e.what() << " of '" << e.get_option_name() << "'." << endl;
+        return -1;
+    }
 #endif
-            cout << "Error: " << e.what() << endl;
-        cout << endl;
-
+    catch (exception& e)
+    {
+        cerr << "Program options invalid due to: " << e.what() << endl;
         return -1;
     }