MONITOR/alistoragemanager/AliStorageClientThread.cxx

   1 #include "AliStorageClientThread.h"
   2 #include "AliMultiplicity.h"
   3 #include "AliStorageTypes.h"
   4 #include "AliStorageEventManager.h"
   5
   6 #include <sstream>
   7 #include <signal.h>
   8 #include <fstream>
   9 #include <iostream>
  10
  11 #include <TSystemDirectory.h>
  12 #include <TThread.h>
  13 #include <TFile.h>
  14
  15 using namespace std;
  16
  17 bool gClientQuit = false;    // signal flag
  18 void GotSignalClient(int){gClientQuit = true;}
  19
  20 AliStorageClientThread::AliStorageClientThread() :
  21 fConnectionStatus(STATUS_WAITING),
  22 fReceivingStatus(STATUS_WAITING),
  23 fSavingStatus(STATUS_WAITING),
  24 fCommunicationThread(0),
  25 fCurrentFile(0),
  26 fDatabase(0),
  27 fCurrentStorageSize(0),
  28 fMaximumStorageSize(0),
  29 fStoragePath(""),
  30 fNumberOfEventsInFile(0),
  31 fStorageOccupationLevel(0),
  32 fRemoveEventsPercentage(0)
  33 {
  34         // make sure that when program is closed destructor will be called
  35         struct sigaction sa;
  36         memset(&sa,0,sizeof(sa));
  37         sa.sa_handler = GotSignalClient;
  38         sigfillset(&sa.sa_mask);
  39         sigaction(SIGINT,&sa,NULL);
  40
  41         //load storage parameters from file
  42         TThread::Lock();
  43         ifstream configFile (GetConfigFilePath());
  44         if (configFile.is_open())
  45         {
  46                 string line;
  47                 int from,to;
  48                 while(configFile.good())
  49                 {
  50                         getline(configFile,line);
  51                         from = line.find("\"")+1;
  52                         to = line.find_last_of("\"");
  53                         if(line.find("STORAGE_PATH=")==0)
  54                         {
  55                                 fStoragePath=line.substr(from,to-from);
  56                         }
  57                         else if(line.find("MAX_SIZE=")==0)
  58                         {
  59                                 fMaximumStorageSize=atoi(line.substr(from,to-from).c_str());
  60                         }
  61                         else if(line.find("MAX_OCCUPATION=")==0)
  62                         {
  63                                 fStorageOccupationLevel=atoi(line.substr(from,to-from).c_str());
  64                         }
  65                         else if(line.find("REMOVE_PERCENT=")==0)
  66                         {
  67                                 fRemoveEventsPercentage=atoi(line.substr(from,to-from).c_str());
  68                         }
  69                         else if(line.find("EVENTS_IN_FILE=")==0)
  70                         {
  71                                 fNumberOfEventsInFile=atoi(line.substr(from,to-from).c_str());
  72                         }
  73                 }
  74                 if(configFile.eof())
  75                 {
  76                         configFile.clear();
  77                 }
  78                 configFile.close();
  79         }
  80         else
  81         {
  82                 cout<<"CLIENT -- Unable to open config file"<<endl;
  83         }
  84         //create directory for storage if it doesn't exist
  85         gSystem->Exec(Form("mkdir -p %s",fStoragePath.c_str()));
  86
  87         //create database class
  88         fDatabase = new AliStorageDatabase();
  89         TThread::UnLock();
  90
  91         //check current storage size
  92         fCurrentStorageSize = GetSizeOfAllChunks();
  93
  94         //create two-way commynication thread
  95         fCommunicationThread = new TThread("fCommunicationThread",
  96                                        Dispatch,(void*)this);
  97         fCommunicationThread->Run();
  98 }
  99
 100 AliStorageClientThread::~AliStorageClientThread()
 101 {
 102         cout<<"CLIENT -- AliStorageClientThread destructor called";
 103         if(fCurrentFile)
 104         {
 105                 fCurrentFile->Close();
 106                 delete fCurrentFile;
 107         }
 108         if(fCommunicationThread){delete fCommunicationThread;}
 109         if(fDatabase){delete fDatabase;}
 110         cout<<" --- OK"<<endl;
 111 }
 112
 113 void AliStorageClientThread::CommunicationHandle()
 114 {
 115         AliStorageEventManager *eventManager = AliStorageEventManager::GetEventManagerInstance();
 116         storageSockets socket = CLIENT_COMMUNICATION_REP;
 117         eventManager->CreateSocket(socket);
 118
 119         struct clientRequestStruct *request;
 120         struct clientRequestStruct *response = new struct clientRequestStruct;
 121
 122         cout<<"CLIENT -- Communication stated"<<endl;
 123
 124         while(!gClientQuit)
 125         {
 126                 request = eventManager->GetClientStruct(socket);
 127                 switch(request->messageType)
 128                 {
 129                 case REQUEST_CONNECTION:
 130                         eventManager->Send((long)fConnectionStatus,socket);
 131                         break;
 132                 case REQUEST_RECEIVING:
 133                         eventManager->Send((long)fReceivingStatus,socket);
 134                         break;
 135                 case REQUEST_SAVING:
 136                         eventManager->Send((long)fSavingStatus,socket);
 137                         break;
 138                 case REQUEST_CURRENT_SIZE:
 139                         eventManager->Send((long)fCurrentStorageSize,socket);
 140                         break;
 141                 case REQUEST_GET_PARAMS:
 142                         response->maxStorageSize = fMaximumStorageSize;
 143                         response->maxOccupation = fStorageOccupationLevel;
 144                         response->removeEvents = fRemoveEventsPercentage;
 145                         response->eventsInChunk = fNumberOfEventsInFile;
 146
 147                         eventManager->Send(response,socket);
 148                         break;
 149                 case REQUEST_SET_PARAMS:
 150                         SetStorageParams(request->maxStorageSize,
 151                                          request->maxOccupation,
 152                                          request->removeEvents,
 153                                          request->eventsInChunk);
 154
 155                         fMaximumStorageSize = request->maxStorageSize;
 156                         fStorageOccupationLevel = request->maxOccupation;
 157                         fRemoveEventsPercentage = request->removeEvents;
 158                         fNumberOfEventsInFile = request->eventsInChunk;
 159
 160                         eventManager->Send(true,socket);
 161                         break;
 162                 default:break;
 163                 }
 164                 delete request;
 165         }
 166 }
 167
 168 void AliStorageClientThread::SetStorageParams(int maxStorageSize,int maxOccupation,int removeEvents,int eventsInChunk)
 169 {
 170         cout<<maxStorageSize<<endl<<maxOccupation<<endl<<removeEvents<<endl<<eventsInChunk<<endl;
 171
 172
 173         TThread::Lock();
 174         ifstream configFile (GetConfigFilePath());
 175         ofstream tmpFile("tmpFile.bla");
 176
 177         if (configFile.is_open())
 178         {
 179                 string line;
 180                 string tmpLine;
 181                 int from,to;
 182                 while(configFile.good())
 183                 {
 184                         getline(configFile,line);
 185                         from = line.find("\"")+1;
 186                         to = line.find_last_of("\"");
 187                         tmpLine = line;
 188                         if(line.find("MAX_SIZE=")==0)
 189                         {
 190                                 tmpLine = Form("MAX_SIZE=\"%d\"",maxStorageSize);
 191                         }
 192                         else if(line.find("MAX_OCCUPATION=")==0)
 193                         {
 194                                 tmpLine = Form("MAX_OCCUPATION=\"%d\"",maxOccupation);
 195                         }
 196                         else if(line.find("REMOVE_PERCENT=")==0)
 197                         {
 198                                 tmpLine = Form("REMOVE_PERCENT=\"%d\"",removeEvents);
 199                         }
 200                         else if(line.find("EVENTS_IN_FILE=")==0)
 201                         {
 202                                 tmpLine = Form("EVENTS_IN_FILE=\"%d\"",eventsInChunk);
 203                         }
 204                         tmpLine += "\n";
 205                         tmpFile << tmpLine;
 206                 }
 207                 if(configFile.eof())
 208                 {
 209                         configFile.clear();
 210                 }
 211                 configFile.close();
 212                 tmpFile.close();
 213                 rename("tmpFile.bla",GetConfigFilePath());
 214         }
 215         else
 216         {
 217                 cout<<"CLIENT -- Unable to open config file"<<endl;
 218         }
 219         TThread::UnLock();
 220 }
 221
 222 Long64_t AliStorageClientThread::GetSizeOfAllChunks()
 223 {
 224         Long64_t totalStorageSize = 0;
 225
 226         TSystemDirectory dir(fStoragePath.c_str(),fStoragePath.c_str());
 227         TList *listOfDirectories = dir.GetListOfFiles();
 228
 229         if (!listOfDirectories)
 230         {
 231                 cout<<"CLIENT -- Storage directory is empty"<<endl;
 232                 return 0;
 233         }
 234         TIter nextDirectory(listOfDirectories);
 235         TSystemFile *runDirectory;
 236         string directoryName;
 237
 238         while ((runDirectory=(TSystemFile*)nextDirectory()))
 239         {
 240                 directoryName=runDirectory->GetName();
 241                 if (runDirectory->IsDirectory() && directoryName.find("run")==0)
 242                 {
 243                         TSystemDirectory dirChunks(Form("%s/%s",fStoragePath.c_str(),directoryName.c_str()),Form("%s/%s",fStoragePath.c_str(),directoryName.c_str()));
 244                         TList *listOfChunks = dirChunks.GetListOfFiles();
 245
 246                         if(listOfChunks)
 247                         {
 248                                 TIter nextChunk(listOfChunks);
 249                                 TSystemFile *chunk;
 250                                 string chunkFileName;
 251
 252                                 while((chunk=(TSystemFile*)nextChunk()))
 253                                 {
 254                                         chunkFileName = chunk->GetName();
 255                                         if(!chunk->IsDirectory() && chunkFileName.find("chunk")==0)
 256                                         {
 257                                                 TFile *tmpFile = new TFile(Form("%s/%s/%s",fStoragePath.c_str(),directoryName.c_str(),chunkFileName.c_str()),"read");
 258                                                 if(tmpFile)
 259                                                 {
 260                                                         totalStorageSize+=tmpFile->GetSize();
 261                                                         tmpFile->Close();
 262                                                         delete tmpFile;
 263                                                 }
 264                                         }
 265                                 }
 266                                 if(chunk){delete chunk;}
 267                         }
 268                         if(listOfChunks){delete listOfChunks;}
 269                 }
 270         }
 271
 272         //tmpFiles.clear();
 273         if(listOfDirectories){delete listOfDirectories;}
 274         if(runDirectory){delete runDirectory;}
 275
 276         printf("CLIENT -- Total storage size:%lld\t(%.2f MB)\n",totalStorageSize,(float)totalStorageSize/(1000.*1000.));
 277
 278         return totalStorageSize;
 279 }
 280
 281 void AliStorageClientThread::CollectData()
 282 {
 283         AliStorageEventManager *eventManager = AliStorageEventManager::GetEventManagerInstance();
 284         if(eventManager->CreateSocket(EVENTS_SERVER_SUB)){fConnectionStatus=STATUS_OK;}
 285         else{fConnectionStatus=STATUS_ERROR;}
 286
 287         int chunkNumber=0;
 288         int previousChunkNumber=-1;
 289         int eventsInChunk=0;
 290         int previousRunNumber=-1;
 291         AliESDEvent *event = NULL;
 292         vector<struct eventStruct> eventsToUpdate;
 293         struct eventStruct currentEvent;
 294
 295
 296         while(!gClientQuit)
 297         {
 298           event = eventManager->GetEvent(EVENTS_SERVER_SUB);
 299
 300                 if(event)
 301                 {
 302                         fReceivingStatus=STATUS_OK;
 303
 304                         if(event->GetRunNumber() != previousRunNumber)//when new run starts
 305                         {
 306                                 cout<<"CLIENT -- new run started"<<endl;
 307                                 previousRunNumber = event->GetRunNumber();
 308                                 gSystem->Exec(Form("mkdir -p %s/run%d",fStoragePath.c_str(),event->GetRunNumber()));
 309                                 chunkNumber=0;
 310                                 eventsInChunk=0;
 311
 312                                 TSystemDirectory dir(Form("%s/run%d",fStoragePath.c_str(),event->GetRunNumber()),
 313                                                      Form("%s/run%d",fStoragePath.c_str(),event->GetRunNumber()));
 314                                 TList *files = dir.GetListOfFiles();
 315                                 if (files)
 316                                 {
 317                                         TSystemFile *file;
 318                                         string fname;
 319                                         TIter next(files);
 320
 321                                         while ((file=(TSystemFile*)next()))
 322                                         {
 323                                                 fname = file->GetName();
 324
 325                                                 if (!file->IsDirectory())
 326                                                 {
 327                                                         int from = fname.find("chunk")+5;
 328                                                         int to = fname.find(".root");
 329
 330                                                         int maxChunkNumber = atoi(fname.substr(from,to-from).c_str());
 331
 332                                                         if(maxChunkNumber > chunkNumber)
 333                                                         {
 334                                                                 chunkNumber = maxChunkNumber;
 335                                                         }
 336                                                 }
 337                                         }
 338                                         chunkNumber++;
 339                                 }
 340                         }
 341
 342                         cout<<"CLIENT -- Received data. Event:"<<event->GetEventNumberInFile()<<"\trun:"<<event->GetRunNumber()<<endl;
 343
 344                         if(chunkNumber != previousChunkNumber)//when new chunk needs to be created
 345                         {
 346                                 if(fCurrentFile)
 347                                 {
 348                                         fCurrentFile->Close();
 349                                         delete fCurrentFile;
 350                                         fCurrentFile=0;
 351                                 }
 352                                 for(unsigned int i=0;i<eventsToUpdate.size();i++)
 353                                   {
 354                                     fDatabase->UpdateEventPath(eventsToUpdate[i],
 355                                                                Form("%s/run%d/chunk%d.root",
 356                                                                     fStoragePath.c_str(),
 357                                                                     event->GetRunNumber(),
 358                                                                     chunkNumber-1));
 359                                   }
 360                                 eventsToUpdate.clear();
 361
 362
 363                                 fCurrentStorageSize=GetSizeOfAllChunks();
 364                                 CheckCurrentStorageSize();
 365
 366                                 fCurrentFile = new TFile(Form("%s/run%d/chunk%d.root", fStoragePath.c_str(),event->GetRunNumber(),chunkNumber),"recreate");
 367
 368                                 previousChunkNumber = chunkNumber;
 369                         }
 370
 371                         //create new directory for this run
 372                         TDirectory *currentRun;
 373                         if((currentRun = fCurrentFile->mkdir(Form("run%d",event->GetRunNumber()))))
 374                           {
 375                             cout<<"CLIENT -- creating new directory for this run"<<endl;
 376                             currentRun->cd();
 377                           }
 378                         else
 379                           {
 380                             cout<<"CLIENT -- opening existing directory for this run"<<endl;
 381                             fCurrentFile->cd(Form("run%d",event->GetRunNumber()));
 382                           }
 383
 384                         if(0 != event->Write(Form("event%d",event->GetEventNumberInFile())))
 385                           //fCurrentFile->WriteObject(event,Form("event%d",event->GetEventNumberInFile())))//if event was written to file
 386                         {
 387                                 eventsInChunk++;
 388
 389                                 if(eventsInChunk == fNumberOfEventsInFile)//if max events number in file was reached
 390                                 {
 391                                         chunkNumber++;
 392                                         eventsInChunk=0;
 393                                 }
 394
 395                                 if(fSavingStatus!=STATUS_OK)
 396                                 {
 397                                         fSavingStatus=STATUS_OK;
 398                                 }
 399                         }
 400                         else if(fSavingStatus!=STATUS_ERROR)
 401                         {
 402                                 fSavingStatus=STATUS_ERROR;
 403                         }
 404
 405                 // save to event file as well:
 406
 407                 TFile *eventFile = new TFile(Form("%s/run%d/event%d.root", fStoragePath.c_str(),event->GetRunNumber(),eventsInChunk),"recreate");
 408
 409                 if((currentRun = eventFile->mkdir(Form("run%d",event->GetRunNumber()))))
 410                           {
 411                             cout<<"CLIENT -- creating new directory for this run"<<endl;
 412                             currentRun->cd();
 413                           }
 414                         else
 415                           {
 416                             cout<<"CLIENT -- opening existing directory for this run"<<endl;
 417                             eventFile->cd(Form("run%d",event->GetRunNumber()));
 418                           }
 419
 420                 if(0 == event->Write(Form("event%d",event->GetEventNumberInFile())) &&
 421                    fSavingStatus!=STATUS_ERROR){fSavingStatus=STATUS_ERROR;}
 422                 else
 423                   {
 424                     eventFile->Close();
 425                     delete eventFile;
 426                     fDatabase->InsertEvent(event->GetRunNumber(),
 427                                            event->GetEventNumberInFile(),
 428                                            (char*)event->GetBeamType(),
 429                                            event->GetMultiplicity()->GetNumberOfTracklets(),
 430                                            Form("%s/run%d/event%d.root",fStoragePath.c_str(),
 431                                                 event->GetRunNumber(),
 432                                                 eventsInChunk));
 433
 434                     currentEvent.runNumber = event->GetRunNumber();
 435                     currentEvent.eventNumber = event->GetEventNumberInFile();
 436                     eventsToUpdate.push_back(currentEvent);
 437                   }
 438                         delete event;event=0;
 439                         //delete tree;
 440                 }
 441                 else if(fReceivingStatus!=STATUS_ERROR)
 442                 {
 443                         cout<<"CLIENT -- ERROR -- NO DATA!"<<endl;
 444                         fReceivingStatus=STATUS_ERROR;
 445                 }
 446         }
 447         if(event){delete event;}
 448 }
 449
 450
 451 void AliStorageClientThread::CheckCurrentStorageSize()
 452 {
 453         if(fCurrentStorageSize >  (float)fStorageOccupationLevel/100. * fMaximumStorageSize)
 454         {
 455                 while(GetSizeOfAllChunks() > (float)fRemoveEventsPercentage/100. * fMaximumStorageSize)
 456                 {
 457                         struct eventStruct oldestEvent = fDatabase->GetOldestEvent();
 458                         string oldestEventPath = fDatabase->GetFilePath(oldestEvent);
 459                         //remove oldest event
 460                         cout<<"CLIENT -- Removing old events:"<<oldestEventPath<<endl;
 461                         gSystem->Exec(Form("rm -f %s",oldestEventPath.c_str()));
 462             fDatabase->RemoveEventsWithPath(oldestEventPath);
 463 //                      fDatabase->RemoveEvent(oldestEvent);
 464                 }
 465         }
 466 }