#ifdef ZMQ
#include "AliStorageEventManager.h"
+#include "AliOnlineReconstructionUtil.h"
+#include "AliGRPPreprocessor.h"
#endif
using std::cout;
TString AliEveEventManager::fgGAliceFileName("galice.root");
TString AliEveEventManager::fgRawFileName("raw.root");
TString AliEveEventManager::fgCdbUri;
+TString AliEveEventManager::fgSpecificCdbUriValue;
+TString AliEveEventManager::fgSpecificCdbUriPath;
TList* AliEveEventManager::fgAODfriends = 0;
fAutoLoadTimerRunning(kFALSE),
fMutex(new TMutex()),
fgSubSock(EVENTS_SERVER_SUB),
+ fCurrentRun(-1),
fEventInUse(1),
fWritingToEventIndex(0),
fIsNewEventAvaliable(false),
+ fOnlineMode(kFALSE),
fStorageDown(false),
fFinished(false)
{
// Constructor with event-id.
-
- InitInternals();
+ if (0 == name.CompareTo("online")) {fOnlineMode = kTRUE;}
+ else{fOnlineMode = kFALSE;}
+
+ InitInternals();
- Open();
- if (ev >= 0)
+ Open();
+ if (ev >= 0)
{
GotoEvent(ev);
}
- if (0 == name.CompareTo("online")) {
- fOnlineMode = kTRUE;
- }
- else{
- fOnlineMode = kFALSE;
- }
-
-
#ifdef ZMQ
+ if(fOnlineMode)
+ {
cout<<"ZMQ FOUND. Starting subscriber thread."<<endl;
fEventListenerThread = new TThread("fEventListenerThread",DispatchEventListener,(void*)this);
fEventListenerThread->Run();
- fStorageManagerWatcherThread = new TThread("fStorageManagerWatcherThread",DispatchStorageManagerWatcher,(void*)this);
- fStorageManagerWatcherThread->Run();
+ // fStorageManagerWatcherThread = new TThread("fStorageManagerWatcherThread",DispatchStorageManagerWatcher,(void*)this);
+ //fStorageManagerWatcherThread->Run();
+ }
#else
cout<<"NO ZMQ FOUND. Online events not avaliable."<<endl;
#endif
// Destructor.
fFinished = true;
-
if(fEventListenerThread)
{
+ fEventListenerThread->Join();
fEventListenerThread->Kill();
delete fEventListenerThread;
+ cout<<"listener thread killed and deleted"<<endl;
}
if(fStorageManagerWatcherThread)
{
+ fStorageManagerWatcherThread->Join();
fStorageManagerWatcherThread->Kill();
delete fStorageManagerWatcherThread;
+ cout<<"storage watcher thread killed and deleted"<<endl;
}
-
fAutoLoadTimer->Stop();
fAutoLoadTimer->Disconnect("Timeout");
fAutoLoadTimer->Disconnect("AutoLoadNextEvent");
- AliEveEventManager *manager = AliEveEventManager::GetCurrent();
- manager->Disconnect("StorageManagerOk");
- manager->Disconnect("StorageManagerDown");
-
if(fSubManagers){delete fSubManagers;}
if(fMutex){delete fMutex;}
-
- if (fIsOpen)
- {
- Close();
- }
+ if (fIsOpen){Close();}
// fTransients->DecDenyDestroy();
// fTransients->Destroy();
void AliEveEventManager::GetNextEvent()
{
#ifdef ZMQ
+ if(!fOnlineMode){return;}
+ cout<<"\n\nGet next event called\n\n"<<endl;
+
AliStorageEventManager *eventManager = AliStorageEventManager::GetEventManagerInstance();
eventManager->CreateSocket(EVENTS_SERVER_SUB);
eventManager->CreateSocket(SERVER_COMMUNICATION_REQ);
list.eventNumber[1]=999999;
list.marked[0]=1;
list.marked[1]=1;
- list.multiplicity[0]=1;
+ list.multiplicity[0]=0;
list.multiplicity[1]=999999;
strcpy(list.system[0],"p-p");
- strcpy(list.system[1],"A-A");
+ strcpy(list.system[1],"");
struct serverRequestStruct *requestMessage = new struct serverRequestStruct;
requestMessage->messageType = REQUEST_LIST_EVENTS;
requestMessage->list = list;
+ cout<<"Sending request for marked events list"<<endl;
eventManager->Send(requestMessage,SERVER_COMMUNICATION_REQ);
- vector<serverListStruct> receivedList = eventManager->GetServerListVector(SERVER_COMMUNICATION_REQ);
-
+ cout<<"Waiting for response"<<endl;
+ vector<serverListStruct> receivedList = eventManager->GetServerListVector(SERVER_COMMUNICATION_REQ,3000);
cout<<"EVENT DISPLAY -- received list of marked events"<<endl;
for(int i=0;i<receivedList.size();i++)
}
int iter=0;
+
+ cout<<"Starting subscriber's loop"<<endl;
while(!fFinished)
{
- if(!fLoopMarked)
+ cout<<"not finished"<<endl;
+ if(!fLoopMarked || receivedList.size()<=0)
{
cout<<"taking event from reco server"<<endl;
- tmpEvent = eventManager->GetEvent(EVENTS_SERVER_SUB,5);
+ tmpEvent = eventManager->GetEvent(EVENTS_SERVER_SUB,5000);
+ cout<<"after get event"<<endl;
if(!tmpEvent){sleep(1);}
}
else
cout<<"taking event from storage manager"<<endl;
if(iter<receivedList.size())
{
+ cout<<"i:"<<iter<<endl;
struct eventStruct mark;
mark.runNumber = receivedList[iter].runNumber;
mark.eventNumber = receivedList[iter].eventNumber;
- struct serverRequestStruct *requestMessage = new struct serverRequestStruct;
requestMessage->messageType = REQUEST_GET_EVENT;
requestMessage->event = mark;
eventManager->Send(requestMessage,SERVER_COMMUNICATION_REQ);
tmpEvent = eventManager->GetEvent(SERVER_COMMUNICATION_REQ);
- delete requestMessage;
+
iter++;
sleep(1);
}
if(tmpEvent)
{
+ cout<<"tmpEvent:"<<tmpEvent->GetRunNumber()<<endl;
if(tmpEvent->GetRunNumber()>=0)
{
fMutex->Lock();
fMutex->UnLock();
}
}
- else
- {
- cout<<"didn't receive new event"<<endl;
- }
+ else{cout<<"didn't receive new event"<<endl;}
+
}
delete requestMessage;
void AliEveEventManager::CheckStorageStatus()
{
#ifdef ZMQ
+ if(!fOnlineMode){return;}
+
AliEveConfigManager *configManager = AliEveConfigManager::GetMaster();
configManager->ConnectEventManagerSignals();
}
sleep(1);
}
+
+ AliEveEventManager *manager = AliEveEventManager::GetCurrent();
+ manager->Disconnect("StorageManagerOk");
+ manager->Disconnect("StorageManagerDown");
+
#endif
}
if ( ! cdb.IsNull()) fgCdbUri = cdb;
}
+void AliEveEventManager::SetSpecificCdbUri(const TString& path,const TString& value)
+{
+ // Set path to specific CDB object, there is no default.
+ if ( ! value.IsNull()) fgSpecificCdbUriValue = value;
+ if ( ! path.IsNull()) fgSpecificCdbUriPath = path;
+}
+
void AliEveEventManager::SetGAliceFileName(const TString& galice)
{
// Set file-name for opening gAlice, default "galice.root".
// Initialize OCDB ... only in master event-manager
- InitOCDB(runNo);
+ InitOCDB(runNo);
fIsOpen = kTRUE;
void AliEveEventManager::InitOCDB(int runNo)
{
- static const TEveException kEH("AliEveEventManager::InitOCDB ");
+ TString cdbPath = Form("local://%s/ed_ocdb_objects/",gSystem->Getenv("HOME"));
+ AliCDBManager* cdb = AliCDBManager::Instance();
+#ifdef ZMQ
+ if(fOnlineMode)
+ {
+ if(runNo != fCurrentRun)
+ {
+ cout<<"Loading OCDB for new run:"<<runNo<<" in online mode."<<endl;
+ TEnv settings;
+ settings.ReadFile(AliOnlineReconstructionUtil::GetPathToServerConf(), kEnvUser);
+ fCurrentRun = runNo;
+ cout<<"config read"<<endl;
+
+ // Retrieve GRP entry for given run from aldaqdb.
+ TString dbHost = settings.GetValue("logbook.host", DEFAULT_LOGBOOK_HOST);
+ Int_t dbPort = settings.GetValue("logbook.port", DEFAULT_LOGBOOK_PORT);
+ TString dbName = settings.GetValue("logbook.db", DEFAULT_LOGBOOK_DB);
+ TString user = settings.GetValue("logbook.user", DEFAULT_LOGBOOK_USER);
+ TString password = settings.GetValue("logbook.pass", DEFAULT_LOGBOOK_PASS);
+
+ gSystem->cd(cdbPath.Data());
+ gSystem->Exec("rm -fr GRP/");
+ cout<<"CDB path for GRP:"<<cdbPath<<endl;
+
+ TString gdc;
+
+ Int_t ret=AliGRPPreprocessor::ReceivePromptRecoParameters(fCurrentRun, dbHost.Data(),
+ dbPort, dbName.Data(),
+ user.Data(), password.Data(),
+ Form("%s",cdbPath.Data()),
+ gdc);
+
+ if(ret>0) Info("RetrieveGRP","Last run of the same type is: %d",ret);
+ else if(ret==0) Warning("RetrieveGRP","No previous run of the same type found");
+ else if(ret<0) Error("Retrieve","Error code while retrieving GRP parameters returned: %d",ret);
+
+
+ cdb->SetDefaultStorage(settings.GetValue("cdb.defaultStorage", DEFAULT_CDB_STORAGE));
+ cdb->SetSpecificStorage("GRP/GRP/Data",cdbPath.Data());
+ cdb->SetRun(fCurrentRun);
+ cdb->Print();
+ }
+ }
+#endif
+
+ static const TEveException kEH("AliEveEventManager::InitOCDB ");
//if (this == fgMaster)
{
- AliCDBManager* cdb = AliCDBManager::Instance();
if (cdb->IsDefaultStorageSet() == kTRUE)
{
Warning(kEH, "CDB already set - using the old storage:\n '%s'",
gEnv->SetValue("Root.Stacktrace", "no");
Fatal("Open()", "OCDB path was not specified.");
}
-
+ cout<<"Setting default storage:"<<fgCdbUri<<endl;
// Handle some special cases for MC (should be in OCDBManager).
if (fgCdbUri == "mcideal://")
cdb->SetDefaultStorage("MC", "Ideal");
cdb->SetDefaultStorage("MC", "Residual");
else if (fgCdbUri == "mcfull://")
cdb->SetDefaultStorage("MC", "Full");
- else if (fgCdbUri == "local://") {
- fgCdbUri = Form("local://%s/OCDB", gSystem->Getenv("ALICE_ROOT"));
+ else if (fgCdbUri == "local://"){
+ fgCdbUri = Form("local://%s/OCDB", gSystem->Getenv("ALICE_ROOT"));
cdb->SetDefaultStorage(fgCdbUri);
- } else
- cdb->SetDefaultStorage(fgCdbUri);
-
+ }
+ else{
+ cdb->SetDefaultStorage(fgCdbUri);
+ }
cdb->SetRun(runNo);
if (cdb->IsDefaultStorageSet() == kFALSE)
throw kEH + "CDB initialization failed for '" + fgCdbUri + "'.";
- }
-
+ }/*
if (fgCdbUri.BeginsWith("local://"))
{
TString curPath = gSystem->WorkingDirectory();
cdb->SetSpecificStorage(grp, lpath);
}
}
- }
+ }*/
}
}
{
// throw (kEH + "Event-loop is under external control.");
#ifdef ZMQ
- if (fStorageDown && -1 == event)
- {
- NextEvent();
- return;
- }
-
- if (fESD)
- {
- // create new server request:
- struct serverRequestStruct *requestMessage = new struct serverRequestStruct;
+ if(fOnlineMode)
+ {
+ if (fStorageDown && -1 == event)
+ {
+ NextEvent();
+ return;
+ }
+
+ if (fESD)
+ {
+ // create new server request:
+ struct serverRequestStruct *requestMessage = new struct serverRequestStruct;
- // set request type:
- if (event == -1) {requestMessage->messageType = REQUEST_GET_LAST_EVENT;}
- else if (event == 0) {requestMessage->messageType = REQUEST_GET_FIRST_EVENT;}
- else if (event == 1) {requestMessage->messageType = REQUEST_GET_PREV_EVENT;}
- else if (event == 2) {requestMessage->messageType = REQUEST_GET_NEXT_EVENT;}
+ // set request type:
+ if (event == -1) {requestMessage->messageType = REQUEST_GET_LAST_EVENT;}
+ else if (event == 0) {requestMessage->messageType = REQUEST_GET_FIRST_EVENT;}
+ else if (event == 1) {requestMessage->messageType = REQUEST_GET_PREV_EVENT;}
+ else if (event == 2) {requestMessage->messageType = REQUEST_GET_NEXT_EVENT;}
- // set event struct:
- struct eventStruct eventToLoad;
- eventToLoad.runNumber = fESD->GetRunNumber();
- eventToLoad.eventNumber = fESD->GetEventNumberInFile();
- requestMessage->event = eventToLoad;
+ // set event struct:
+ struct eventStruct eventToLoad;
+ eventToLoad.runNumber = fESD->GetRunNumber();
+ eventToLoad.eventNumber = fESD->GetEventNumberInFile();
+ requestMessage->event = eventToLoad;
- // create event manager:
- AliStorageEventManager *eventManager =
- AliStorageEventManager::GetEventManagerInstance();
- AliESDEvent *resultEvent = NULL;
+ // create event manager:
+ AliStorageEventManager *eventManager =
+ AliStorageEventManager::GetEventManagerInstance();
+ AliESDEvent *resultEvent = NULL;
- eventManager->CreateSocket(SERVER_COMMUNICATION_REQ);
- fMutex->Lock();
+ eventManager->CreateSocket(SERVER_COMMUNICATION_REQ);
+ fMutex->Lock();
- // send request and receive event:
- eventManager->Send(requestMessage,SERVER_COMMUNICATION_REQ);
- resultEvent = eventManager->GetEvent(SERVER_COMMUNICATION_REQ);
+ // send request and receive event:
+ eventManager->Send(requestMessage,SERVER_COMMUNICATION_REQ);
+ resultEvent = eventManager->GetEvent(SERVER_COMMUNICATION_REQ);
- if(resultEvent)
- {
- DestroyElements();
- InitOCDB(resultEvent->GetRunNumber());
- SetEvent(0,0,resultEvent,0);
- }
- else
- {
- if(event==-1){cout<<"\n\nWARNING -- No last event is avaliable.\n\n"<<endl;}
- if(event==0){cout<<"\n\nWARNING -- No first event is avaliable.\n\n"<<endl;}
- if(event==1){cout<<"\n\nWARNING -- No previous event is avaliable.\n\n"<<endl;}
- if(event==2){cout<<"\n\nWARNING -- No next event is avaliable.\n\n"<<endl;}
- }
+ if(resultEvent)
+ {
+ DestroyElements();
+ InitOCDB(resultEvent->GetRunNumber());
+ SetEvent(0,0,resultEvent,0);
+ }
+ else
+ {
+ if(event==-1){cout<<"\n\nWARNING -- No last event is avaliable.\n\n"<<endl;}
+ if(event==0){cout<<"\n\nWARNING -- No first event is avaliable.\n\n"<<endl;}
+ if(event==1){cout<<"\n\nWARNING -- No previous event is avaliable.\n\n"<<endl;}
+ if(event==2){cout<<"\n\nWARNING -- No next event is avaliable.\n\n"<<endl;}
+ }
- fMutex->UnLock();
- }
- else
- {
- cout<<"\n\nWARNING -- No event has been already loaded. Loading the most recent event...\n\n"<<endl;
+ fMutex->UnLock();
+ }
+ else
+ {
+ cout<<"\n\nWARNING -- No event has been already loaded. Loading the most recent event...\n\n"<<endl;
- struct serverRequestStruct *requestMessage = new struct serverRequestStruct;
- requestMessage->messageType = REQUEST_GET_LAST_EVENT;
+ struct serverRequestStruct *requestMessage = new struct serverRequestStruct;
+ requestMessage->messageType = REQUEST_GET_LAST_EVENT;
- AliStorageEventManager *eventManager = AliStorageEventManager::GetEventManagerInstance();
- eventManager->CreateSocket(SERVER_COMMUNICATION_REQ);
- AliESDEvent *resultEvent = NULL;
+ AliStorageEventManager *eventManager = AliStorageEventManager::GetEventManagerInstance();
+ eventManager->CreateSocket(SERVER_COMMUNICATION_REQ);
+ AliESDEvent *resultEvent = NULL;
- fMutex->Lock();
- eventManager->Send(requestMessage,SERVER_COMMUNICATION_REQ);
- resultEvent = eventManager->GetEvent(SERVER_COMMUNICATION_REQ);
+ fMutex->Lock();
+ eventManager->Send(requestMessage,SERVER_COMMUNICATION_REQ);
+ resultEvent = eventManager->GetEvent(SERVER_COMMUNICATION_REQ);
- if(resultEvent)
- {
- fESD=resultEvent;
- DestroyElements();
- InitOCDB(resultEvent->GetRunNumber());
- SetEvent(0,0,resultEvent,0);
- }
- else{cout<<"\n\nWARNING -- The most recent event is not avaliable.\n\n"<<endl;}
- fMutex->UnLock();
- }
+ if(resultEvent)
+ {
+ fESD=resultEvent;
+ DestroyElements();
+ InitOCDB(resultEvent->GetRunNumber());
+ SetEvent(0,0,resultEvent,0);
+ }
+ else{cout<<"\n\nWARNING -- The most recent event is not avaliable.\n\n"<<endl;}
+ fMutex->UnLock();
+ }
+ }
#endif
}
{
throw (kEH + "neither RunLoader, ESD nor Raw loaded.");
}
- if (event < 0 || event > maxEvent)
+ if (event < 0)
{
throw (kEH + Form("event %d not present, available range [%d, %d].",
event, 0, maxEvent));
}
+ if (event > maxEvent)
+ {
+ event=0;
+ cout<<"Event number out of range. Going to event 0"<<endl;
+ }
TString sysInfoHeader;
sysInfoHeader.Form("AliEveEventManager::GotoEvent(%d) - ", event);
if (fExternalCtrl)
{
#ifdef ZMQ
- if(fIsNewEventAvaliable)
- {
- fMutex->Lock();
- if(fWritingToEventIndex == 0) fEventInUse = 0;
- else if(fWritingToEventIndex == 1) fEventInUse = 1;
+ if(fOnlineMode)
+ {
+ if(fIsNewEventAvaliable)
+ {
+ fMutex->Lock();
+ if(fWritingToEventIndex == 0) fEventInUse = 0;
+ else if(fWritingToEventIndex == 1) fEventInUse = 1;
- if(fCurrentEvent[fEventInUse])
- {
- if(fCurrentEvent[fEventInUse]->GetRunNumber() >= 0)
- {
- printf("======================= setting event to %d\n", fEventId);
+ if(fCurrentEvent[fEventInUse])
+ {
+ if(fCurrentEvent[fEventInUse]->GetRunNumber() >= 0)
+ {
+ printf("======================= setting event to %d\n", fCurrentEvent[fEventInUse]->GetEventNumberInFile());
- DestroyElements();
- InitOCDB(fCurrentEvent[fEventInUse]->GetRunNumber());
- SetEvent(0,0,fCurrentEvent[fEventInUse],0);
+ DestroyElements();
+ InitOCDB(fCurrentEvent[fEventInUse]->GetRunNumber());
+ SetEvent(0,0,fCurrentEvent[fEventInUse],0);
- }
- }
- fIsNewEventAvaliable = false;
- fMutex->UnLock();
- }
- else
- {
- cout<<"No new event is avaliable."<<endl;
- NoEventLoaded();
- }
+ }
+ }
+ fIsNewEventAvaliable = false;
+ fMutex->UnLock();
+ }
+ else
+ {
+ cout<<"No new event is avaliable."<<endl;
+ NoEventLoaded();
+ }
+ }
#endif
}
else if ((fESDTree!=0) || (fHLTESDTree!=0))
void AliEveEventManager::MarkCurrentEvent()
{
#ifdef ZMQ
+ if(!fOnlineMode){return;}
+
struct serverRequestStruct *requestMessage = new struct serverRequestStruct;
struct eventStruct mark;
mark.runNumber = fESD->GetRunNumber();
// Close the event data-files and delete ESD, ESDfriend, run-loader
// and raw-reader.
+ cout<<"\n\n\nClose() called!!\n\n\n"<<endl;
+
static const TEveException kEH("AliEveEventManager::Close ");
if (!fIsOpen)
void AliEveEventManager::SetLoopMarked(Bool_t loopMarked)
{
// Set the automatic event loading mode
- fLoopMarked = loopMarked;
- if (fLoopMarked){StartAutoLoadTimer();}
- else{StopAutoLoadTimer();}
+ if(fLoopMarked == loopMarked)
+ {
+ cout<<"loop marked has the same value - ignoring"<<endl;
+ return;
+ }
+ else
+ {
+ cout<<"Setting loof marked to:"<<loopMarked<<endl;
+ fLoopMarked = loopMarked;
+ if (fLoopMarked){StartAutoLoadTimer();}
+ else{StopAutoLoadTimer();}
+ }
}
void AliEveEventManager::SetTrigSel(Int_t trig)
git://git.uio.no / u / mrichter / AliRoot.git / blobdiff