#include <TEnv.h>
#include <TString.h>
#include <TMap.h>
+#include <TROOT.h>
#ifdef ZMQ
#include "AliStorageEventManager.h"
+#include "AliOnlineReconstructionUtil.h"
+#include "AliGRPPreprocessor.h"
#endif
+using std::cout;
+using std::endl;
+using std::vector;
//==============================================================================
//==============================================================================
// AliEveEventManager
TString AliEveEventManager::fgGAliceFileName("galice.root");
TString AliEveEventManager::fgRawFileName("raw.root");
TString AliEveEventManager::fgCdbUri;
+TString AliEveEventManager::fgSpecificCdbUriValue;
+TString AliEveEventManager::fgSpecificCdbUriPath;
TList* AliEveEventManager::fgAODfriends = 0;
fESDfriend (0), fESDfriendExists(kFALSE),
fAODFile (0), fAODTree (0), fAOD (0),
fRawReader (0), fEventInfo(),
- fAutoLoad (kFALSE), fAutoLoadTime (5), fAutoLoadTimer(0),
+ fAutoLoad (kFALSE),fLoopMarked(kFALSE), fAutoLoadTime (5), fAutoLoadTimer(0),
fIsOpen (kFALSE), fHasEvent (kFALSE), fExternalCtrl (kFALSE),
fGlobal (0), fGlobalReplace (kTRUE), fGlobalUpdate (kTRUE),
fExecutor (0), fTransients(0), fTransientLists(0),
fPEventSelector(0),
fSubManagers (0),
fAutoLoadTimerRunning(kFALSE),
+ fMutex(new TMutex()),
fgSubSock(EVENTS_SERVER_SUB),
+ fCurrentRun(-1),
fEventInUse(1),
fWritingToEventIndex(0),
- fIsNewEventAvaliable(false)
+ 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);
}
#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
AliEveEventManager::~AliEveEventManager()
{
// 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");
- delete fSubManagers;
+ if(fSubManagers){delete fSubManagers;}
+ if(fMutex){delete fMutex;}
+ if (fIsOpen){Close();}
- if (fIsOpen)
- {
- Close();
- }
-
- fTransients->DecDenyDestroy();
- fTransients->Destroy();
+// fTransients->DecDenyDestroy();
+// fTransients->Destroy();
- fTransientLists->DecDenyDestroy();
- fTransientLists->Destroy();
+// fTransientLists->DecDenyDestroy();
+// fTransientLists->Destroy();
//delete fExecutor;
}
void AliEveEventManager::GetNextEvent()
{
#ifdef ZMQ
- AliStorageEventManager *eventManager =
- AliStorageEventManager::GetEventManagerInstance();
+ 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);
fCurrentEvent[0]=0;
fCurrentEvent[1]=0;
- fCurrentTree[0]=0;
- fCurrentTree[1]=0;
+
AliESDEvent *tmpEvent = NULL;
- TTree *tmpTree = NULL;
- while(1)
+ // get list of marked events:
+ struct listRequestStruct list;
+
+ list.runNumber[0]=0;
+ list.runNumber[1]=999999;
+ list.eventNumber[0]=0;
+ list.eventNumber[1]=999999;
+ list.marked[0]=1;
+ list.marked[1]=1;
+ list.multiplicity[0]=0;
+ list.multiplicity[1]=999999;
+ strcpy(list.system[0],"p-p");
+ 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);
+ 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++)
{
- //if(tmpEvent){delete tmpEvent;tmpEvent=0;}
- tmpEvent = eventManager->GetEvent(EVENTS_SERVER_SUB,-1,&tmpTree);
+ cout<<"ev:"<<receivedList[i].eventNumber<<endl;
+ }
+
+ int iter=0;
+
+ cout<<"Starting subscriber's loop"<<endl;
+ while(!fFinished)
+ {
+ cout<<"not finished"<<endl;
+ if(!fLoopMarked || receivedList.size()<=0)
+ {
+ cout<<"taking event from reco server"<<endl;
+ 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;
+
+ requestMessage->messageType = REQUEST_GET_EVENT;
+ requestMessage->event = mark;
+
+ eventManager->Send(requestMessage,SERVER_COMMUNICATION_REQ);
+ tmpEvent = eventManager->GetEvent(SERVER_COMMUNICATION_REQ);
+
+ iter++;
+ sleep(1);
+ }
+ else{iter=0;}
+ }
if(tmpEvent)
{
+ cout<<"tmpEvent:"<<tmpEvent->GetRunNumber()<<endl;
if(tmpEvent->GetRunNumber()>=0)
{
- fMutex.Lock();
+ fMutex->Lock();
if(fEventInUse == 0){fWritingToEventIndex = 1;}
else if(fEventInUse == 1){fWritingToEventIndex = 0;}
- cout<<"Writing to:"<<fWritingToEventIndex<<endl;
+ cout<<"Received new event:"<<tmpEvent->GetEventNumberInFile()<<endl;
if(fCurrentEvent[fWritingToEventIndex])
{
- cout<<"DELETING:"<<fCurrentEvent[fWritingToEventIndex]<<endl;
delete fCurrentEvent[fWritingToEventIndex];
fCurrentEvent[fWritingToEventIndex]=0;
- delete fCurrentTree[fWritingToEventIndex];
}
fCurrentEvent[fWritingToEventIndex] = tmpEvent;
- fCurrentTree[fWritingToEventIndex] = tmpTree;
fIsNewEventAvaliable = true;
- fMutex.UnLock();
+ NewEventLoaded();
+ fMutex->UnLock();
}
}
+ else{cout<<"didn't receive new event"<<endl;}
+
}
+ delete requestMessage;
+
#endif
}
void AliEveEventManager::CheckStorageStatus()
{
#ifdef ZMQ
+ if(!fOnlineMode){return;}
+
AliEveConfigManager *configManager = AliEveConfigManager::GetMaster();
configManager->ConnectEventManagerSignals();
struct clientRequestStruct *request = new struct clientRequestStruct;
request->messageType = REQUEST_CONNECTION;
- while (1)
+ while (!fFinished)
{
if(eventManager->Send(request,CLIENT_COMMUNICATION_REQ,5000))
{
StorageManagerOk();
- cout<<"WARNING -- Storage Manager is OK."<<endl;
long response = eventManager->GetLong(CLIENT_COMMUNICATION_REQ);
- cout<<"RESPONSE:"<<response<<endl;
+ fStorageDown = kFALSE;
}
else
{
StorageManagerDown();
cout<<"WARNING -- Storage Manager is DOWN!!"<<endl;
+ fStorageDown = kTRUE;
}
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);
}
}
- }
+ }*/
}
}
AfterNewEventLoaded();
- if (fAutoLoad) StartAutoLoadTimer();
+ if (fAutoLoad || fLoopMarked) StartAutoLoadTimer();
}
void AliEveEventManager::GotoEvent(Int_t event)
{
+ cout<<"Go to event:"<<event<<endl;
// Load data for specified event.
// If event is out of range an exception is thrown and old state
// is preserved.
}
if (fExternalCtrl)
{
- throw (kEH + "Event-loop is under external control.");
+ // throw (kEH + "Event-loop is under external control.");
+#ifdef ZMQ
+ 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 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;
+
+ eventManager->CreateSocket(SERVER_COMMUNICATION_REQ);
+ fMutex->Lock();
+
+ // 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;}
+ }
+
+ 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;
+
+ 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);
+
+ 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
+
}
else if (!fIsOpen)
{
{
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);
void AliEveEventManager::PrepareForNewEvent(AliESDEvent *event)
{
DestroyElements();
-
InitOCDB(event->GetRunNumber());
-
printf("======================= setting event to %d\n", fEventId);
SetEvent(0,0,event,0);
}
static const TEveException kEH("AliEveEventManager::NextEvent ");
- if (fAutoLoadTimerRunning)
- {
- throw (kEH + "Event auto-load timer is running.");
- }
-
+ if (fAutoLoadTimerRunning){throw (kEH + "Event auto-load timer is running.");}
+
if (fExternalCtrl)
{
#ifdef ZMQ
-
- cout<<fIsNewEventAvaliable<<"\t"<<"\t"<<fWritingToEventIndex<<endl;
-
- if(fIsNewEventAvaliable)
- {
- cout<<"new event"<<endl;
- fMutex.Lock();
- if(fWritingToEventIndex == 0) fEventInUse = 0;
- else if(fWritingToEventIndex == 1) fEventInUse = 1;
- cout<<"Using:"<<fEventInUse<<endl;
-
- if(fCurrentEvent[fEventInUse])
- {
- if(fCurrentEvent[fEventInUse]->GetRunNumber() >= 0)
- {
- printf("======================= setting event to %d\n", fEventId);
-
- DestroyElements();
- InitOCDB(fCurrentEvent[fEventInUse]->GetRunNumber());
- SetEvent(0,0,fCurrentEvent[fEventInUse],0);
- }
- }
- fIsNewEventAvaliable = false;
- fMutex.UnLock();
- }
- else{cout<<"No new event is avaliable."<<endl;}
-
-#endif
+ 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", fCurrentEvent[fEventInUse]->GetEventNumberInFile());
+
+ DestroyElements();
+ InitOCDB(fCurrentEvent[fEventInUse]->GetRunNumber());
+ SetEvent(0,0,fCurrentEvent[fEventInUse],0);
+
+ }
+ }
+ 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
+ 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)
{
static const TEveException kEH("AliEveEventManager::SetTrigSel ");
StopAutoLoadTimer();
NextEvent();
- if (fAutoLoad)
+ if (fAutoLoad || fLoopMarked)
StartAutoLoadTimer();
}
// Emit NewEventLoaded signal.
Emit("NewEventLoaded()");
}
+void AliEveEventManager::NoEventLoaded()
+{
+ // Emit NoEventLoaded signal.
+ Emit("NoEventLoaded()");
+}
void AliEveEventManager::StorageManagerOk()
{
// Emit StorageManagerOk signal.
{
return dynamic_cast<TEveElement*>(fGlobal->GetValue(tag));
}
+Int_t AliEveEventManager::NewEventAvailable()
+{
+ if (fIsNewEventAvaliable)
+ {
+ return 1;
+ }
+ else
+ {
+ return 0;
+ }
+}
git://git.uio.no / u / mrichter / AliRoot.git / blobdiff