MONITOR/alistoragemanager/AliCommunicationThread.cxx

   1 #include "AliCommunicationThread.h"
   2 #include "AliStorageEventManager.h"
   3
   4 #include <iostream>
   5 #include <fstream>
   6
   7 using namespace std;
   8
   9 AliCommunicationThread::AliCommunicationThread(AliStorageClientThread *onlineReconstructionManager) :
  10     fFinished(false),
  11     fManager(onlineReconstructionManager),
  12     fCommunicationThread(0)
  13 {
  14     //create two-way communication thread
  15     fCommunicationThread = new TThread("fCommunicationThread",Dispatch,(void*)this);
  16     fCommunicationThread->Run();
  17 }
  18
  19 AliCommunicationThread::~AliCommunicationThread()
  20 {
  21     if(fCommunicationThread){delete fCommunicationThread;}
  22 }
  23
  24 void AliCommunicationThread::Kill()
  25 {
  26     if(fCommunicationThread)
  27     {
  28         fFinished=true;
  29         fCommunicationThread->Join();
  30         fCommunicationThread->Kill();
  31     }
  32 }
  33
  34 void AliCommunicationThread::CommunicationHandle()
  35 {
  36     AliStorageEventManager *eventManager = AliStorageEventManager::GetEventManagerInstance();
  37     storageSockets socket = CLIENT_COMMUNICATION_REP;
  38     eventManager->CreateSocket(socket);
  39
  40     struct clientRequestStruct *request;
  41     struct clientRequestStruct *response = new struct clientRequestStruct;
  42
  43     cout<<"CLIENT -- Communication stated"<<endl;
  44
  45     mutex mtx;
  46
  47     while(!fFinished)
  48     {
  49         cout<<"COMMUNICATION -- waiting for requests"<<endl;
  50         request = eventManager->GetClientStruct(socket,5000);
  51
  52         if(request)
  53         {
  54             lock_guard<mutex> lock(mtx);
  55             cout<<"COMMUNICATION -- received request"<<endl;
  56             switch(request->messageType)
  57             {
  58                 case REQUEST_CONNECTION:
  59                     eventManager->Send((long)fManager->fConnectionStatus,socket);
  60                     break;
  61                 case REQUEST_RECEIVING:
  62                     eventManager->Send((long)fManager->fReceivingStatus,socket);
  63                     break;
  64                 case REQUEST_SAVING:
  65                     eventManager->Send((long)fManager->fSavingStatus,socket);
  66                     break;
  67                 case REQUEST_CURRENT_SIZE:
  68                     eventManager->Send((long)fManager->fCurrentStorageSize,socket);
  69                     break;
  70                 case REQUEST_GET_PARAMS:
  71                     response->maxStorageSize = fManager->fMaximumStorageSize;
  72                     response->maxOccupation = fManager->fStorageOccupationLevel;
  73                     response->removeEvents = fManager->fRemoveEventsPercentage;
  74                     response->eventsInChunk = fManager->fNumberOfEventsInFile;
  75
  76                     eventManager->Send(response,socket);
  77                     break;
  78                 case REQUEST_SET_PARAMS:
  79                     SetStorageParams(request->maxStorageSize,
  80                                      request->maxOccupation,
  81                                      request->removeEvents,
  82                                      request->eventsInChunk);
  83
  84                     fManager->fMaximumStorageSize = request->maxStorageSize;
  85                     fManager->fStorageOccupationLevel = request->maxOccupation;
  86                     fManager->fRemoveEventsPercentage = request->removeEvents;
  87                     fManager->fNumberOfEventsInFile = request->eventsInChunk;
  88
  89                     eventManager->Send(true,socket);
  90                     break;
  91                 default:break;
  92             }
  93             delete request;
  94         }
  95         else{cout<<"COMMUNICATION -- received NO request"<<endl;}
  96     }
  97 }
  98
  99 void AliCommunicationThread::SetStorageParams(int maxStorageSize,int maxOccupation,int removeEvents,int eventsInChunk)
 100 {
 101     cout<<maxStorageSize<<endl<<maxOccupation<<endl<<removeEvents<<endl<<eventsInChunk<<endl;
 102
 103     TThread::Lock();
 104     ifstream configFile (GetConfigFilePath());
 105     ofstream tmpFile("tmpFile.bla");
 106
 107     if (configFile.is_open())
 108     {
 109         string line;
 110         string tmpLine;
 111         int from,to;
 112         while(configFile.good())
 113         {
 114             getline(configFile,line);
 115             from = line.find("\"")+1;
 116             to = line.find_last_of("\"");
 117             tmpLine = line;
 118             if(line.find("MAX_SIZE=")==0){
 119                 tmpLine = Form("MAX_SIZE=\"%d\"",maxStorageSize);
 120             }
 121             else if(line.find("MAX_OCCUPATION=")==0){
 122                 tmpLine = Form("MAX_OCCUPATION=\"%d\"",maxOccupation);
 123             }
 124             else if(line.find("REMOVE_PERCENT=")==0){
 125                 tmpLine = Form("REMOVE_PERCENT=\"%d\"",removeEvents);
 126             }
 127             else if(line.find("EVENTS_IN_FILE=")==0){
 128                 tmpLine = Form("EVENTS_IN_FILE=\"%d\"",eventsInChunk);
 129             }
 130             tmpLine += "\n";
 131             tmpFile << tmpLine;
 132         }
 133         if(configFile.eof()){configFile.clear();}
 134         configFile.close();
 135         tmpFile.close();
 136         rename("tmpFile.bla",GetConfigFilePath());
 137     }
 138     else{cout<<"CLIENT -- Unable to open config file"<<endl;}
 139     TThread::UnLock();
 140 }