MONITOR/alieventserver/AliEventServer.cxx

   1 // Author:  Mihai Niculesu 2013
   2
   3 /**************************************************************************
   4  * Copyright(c) 1998-2013, ALICE Experiment at CERN, all rights reserved. *)
   5  * See http://aliceinfo.cern.ch/Offline/AliRoot/License.html for          *
   6  * full copyright notice.                                                 *
   7  **************************************************************************/
   8
   9 #include <TEnv.h>
  10 #include <TSystem.h>
  11
  12 #include <TSQLServer.h>
  13 #include <TSQLResult.h>
  14 #include <TSQLRow.h>
  15
  16 #include <TTimeStamp.h>
  17 #include <TTimer.h>
  18 #include <iostream>
  19
  20 #include <AliLog.h>
  21 #include <AliReconstruction.h>
  22 #include <AliDimIntNotifier.h>
  23
  24 #include "AliEventServerUtil.h"
  25 #include "AliEventServer.h"
  26 #include "AliEventServerReconstruction.h"
  27
  28 ClassImp(AliEventServer)
  29
  30 using namespace std;
  31
  32 AliEventServer::AliEventServer() :
  33         fRecoServer(0)
  34 {
  35         fRecoServer = new AliEventServerReconstruction();
  36         for(Int_t i=0; i<5; ++i)
  37         {
  38                 fDimSORListener[i] = 0;
  39                 fDimEORListener[i] = 0;
  40         }
  41         FillRunsFromDatabase();
  42         InitDIMListeners();
  43 }
  44
  45 AliEventServer::~AliEventServer()
  46 {
  47         for (Int_t i = 0; i < 5; ++i)
  48         {
  49                 if(fDimSORListener[i]) delete fDimSORListener[i];
  50                 if(fDimEORListener[i]) delete fDimEORListener[i];
  51
  52                 fDimSORListener[i] = 0;
  53                 fDimEORListener[i] = 0;
  54         }
  55         if(fRecoServer){delete fRecoServer;fRecoServer=0;}
  56 }
  57
  58 void AliEventServer::InitDIMListeners()
  59 {
  60         // DIM interface.
  61         for (Int_t i = 0; i < 5; ++i)
  62         {
  63                 if (i == 0)
  64                 {
  65                         fDimSORListener[i] = new AliDimIntNotifier("/LOGBOOK/SUBSCRIBE/DAQ_SOR_PHYSICS");
  66                         fDimEORListener[i] = new AliDimIntNotifier("/LOGBOOK/SUBSCRIBE/DAQ_EOR_PHYSICS");
  67                 }
  68                 else
  69                 {
  70                         fDimSORListener[i] = new AliDimIntNotifier(Form("/LOGBOOK/SUBSCRIBE/DAQ_SOR_PHYSICS_%d", i));
  71                         fDimEORListener[i] = new AliDimIntNotifier(Form("/LOGBOOK/SUBSCRIBE/DAQ_EOR_PHYSICS_%d", i));
  72                 }
  73
  74                 fDimSORListener[i]->Connect("DimMessage(Int_t)", "AliEventServer", this, "StartOfRun(Int_t)");
  75                 fDimEORListener[i]->Connect("DimMessage(Int_t)", "AliEventServer", this, "EndOfRun(Int_t)");
  76         }
  77
  78 }
  79
  80 void AliEventServer::StartOfRun(Int_t run)
  81 {
  82   cout<<"SOR signal received for run:"<<run<<endl;
  83   if(run<=0) return;
  84   fRecoServer->StopReconstruction();
  85
  86   TEnv settings;
  87   settings.ReadFile(AliEventServerUtil::GetPathToServerConf(), kEnvUser);
  88
  89   TString dataSource = settings.GetValue("data.source", DEFAULT_DATA_SOURCE);
  90   TString eventSource;
  91
  92   if(dataSource=="local")
  93     {
  94       cout<<"Starting Reco for run "<<run<<endl;
  95       eventSource = Form("mem://%s/run%d", gSystem->Getenv("ONLINERECO_RAWFILES_DIR"), run);
  96     }
  97   else if(dataSource=="run")
  98     {
  99       cout<<"Starting Reco for GDCs active in current run:"<<run<<endl;
 100       eventSource = "mem://@*:";
 101     }
 102
 103   fRecoServer->StartReconstruction(run, eventSource.Data());
 104 }
 105
 106 void AliEventServer::EndOfRun(Int_t run)
 107 {
 108   cout<<"EOR signal received for run:"<<run<<endl;
 109   if(run<=0) return;
 110   fRecoServer->StopReconstruction();
 111 }
 112
 113 void AliEventServer::FillRunsFromDatabase()
 114 {
 115         TEnv settings;
 116         settings.ReadFile(AliEventServerUtil::GetPathToServerConf(), kEnvUser);
 117
 118         TString dbHost = settings.GetValue("logbook.host", DEFAULT_LOGBOOK_HOST);
 119         TString dbPort =  Form("%d", settings.GetValue("logbook.port", DEFAULT_LOGBOOK_PORT));
 120         TString dbName =  settings.GetValue("logbook.db", DEFAULT_LOGBOOK_DB);
 121         TString user =  settings.GetValue("logbook.user", DEFAULT_LOGBOOK_USER);
 122         TString password = settings.GetValue("logbook.pass", DEFAULT_LOGBOOK_PASS);
 123
 124         TString connStr = Form("mysql://%s:%s/%s", dbHost.Data(), dbPort.Data(), dbName.Data()) ;
 125
 126         cout<<"connecting to:"<<connStr<<endl;
 127
 128         TSQLServer* server = TSQLServer::Connect(connStr.Data(), user.Data(), password.Data());
 129         if (!server)
 130         {
 131           cout<<"ERROR: Could not connect to DAQ Logbook"<<endl;
 132                 return;
 133         }
 134         TString sqlQuery;
 135         TTimeStamp ts;
 136         sqlQuery.Form("SELECT run FROM logbook WHERE DAQ_time_start > %u AND DAQ_time_end IS NULL AND `partition` REGEXP 'PHYSICS.*'",
 137                       (UInt_t)ts.GetSec()-86400);
 138         TSQLResult* result = server->Query(sqlQuery);
 139         if (!result)
 140         {
 141           cout<<"ERROR: Can't execute query:"<< sqlQuery<<endl;
 142                 return;
 143         }
 144         if (result->GetRowCount() != 0)
 145         {
 146                 for (Int_t iRow = 0; iRow < result->GetRowCount(); iRow++)
 147                 {
 148                         TSQLRow* row = result->Next();
 149                         TString runStr = row->GetField(0);
 150                         if (runStr.IsDigit())
 151                                 StartOfRun(runStr.Atoi());
 152                         delete row;
 153                 }
 154         }
 155         delete result;
 156
 157 }
 158
 159
 160 /*
 161 void AliEventServer::FinishedReconstruction(Int_t status)
 162 {
 163   // Slot called on termination of child process.
 164   Int_t run = fServer->GetRunId();
 165
 166   Info("FinishedReconstruction", "Reconstruction Thread finished \tRunId:%d \tstatus=%d", run, status);
 167
 168   mIntInt_i i =fRun2PidMap.find(run);
 169   if (i != fRun2PidMap.end())
 170     {
 171       fRunList->RemoveEntry(run);
 172
 173       // clean (remove) run's reconstructed directory
 174       //gSystem->Exec(Form("rm -rf %s/reco/run%d_%d",gSystem->Getenv("ONLINERECO_BASE_DIR"),run,pid));
 175
 176       if (status == 0)
 177         {
 178           fRunList->AddEntrySort(TString::Format("%-20d -- PROCESSED", run), run);
 179         }
 180       else
 181         {
 182           fRunList->AddEntrySort(TString::Format("%-20d -- PROCESSED [%d]", run, status), run);
 183         }
 184       fRunList->Layout();
 185
 186     }
 187   else
 188     {
 189       Warning("FinishedReconstruction", "Run number %d not registered.", run);
 190     }
 191
 192     }*/