[u/mrichter/AliRoot.git] / MONITOR / alistoragemanager / AliStorageClientThread.cxx

#include "AliStorageClientThread.h"
#include "AliMultiplicity.h"
#include "AliStorageTypes.h"
#include "AliStorageEventManager.h"

#include <sstream>
#include <signal.h>
#include <fstream>
#include <iostream>

#include <TSystemDirectory.h>
#include <TThread.h>
#include <TFile.h>

using namespace std;

bool gClientQuit = false;    // signal flag
void GotSignalClient(int){gClientQuit = true;}

AliStorageClientThread::AliStorageClientThread() :
fConnectionStatus(STATUS_WAITING),
fReceivingStatus(STATUS_WAITING),
fSavingStatus(STATUS_WAITING),
fCommunicationThread(0),
fCurrentFile(0),
fDatabase(0),
fCurrentStorageSize(0),
fMaximumStorageSize(0),
fStoragePath(""),
fNumberOfEventsInFile(0),
fStorageOccupationLevel(0),
fRemoveEventsPercentage(0)
{	
	// make sure that when program is closed destructor will be called
	struct sigaction sa;
	memset(&sa,0,sizeof(sa));
	sa.sa_handler = GotSignalClient;
	sigfillset(&sa.sa_mask);
	sigaction(SIGINT,&sa,NULL);
	
	//load storage parameters from file
	TThread::Lock();
	ifstream configFile (GetConfigFilePath());
	if (configFile.is_open())
	{
		string line;
		int from,to;
		while(configFile.good())
		{
			getline(configFile,line);
			from = line.find("\"")+1;
			to = line.find_last_of("\"");
			if(line.find("STORAGE_PATH=")==0)
			{
				fStoragePath=line.substr(from,to-from);
			}
			else if(line.find("MAX_SIZE=")==0)
			{
				fMaximumStorageSize=atoi(line.substr(from,to-from).c_str());
			}
			else if(line.find("MAX_OCCUPATION=")==0)
			{
				fStorageOccupationLevel=atoi(line.substr(from,to-from).c_str());
			}
			else if(line.find("REMOVE_PERCENT=")==0)
			{
				fRemoveEventsPercentage=atoi(line.substr(from,to-from).c_str());
			}
			else if(line.find("EVENTS_IN_FILE=")==0)
			{
				fNumberOfEventsInFile=atoi(line.substr(from,to-from).c_str());
			}
		}
		if(configFile.eof())
		{
			configFile.clear();
		}
		configFile.close();
	}
	else
	{
		cout<<"CLIENT -- Unable to open config file"<<endl;
	}
	//create directory for storage if it doesn't exist
	gSystem->Exec(Form("mkdir -p %s",fStoragePath.c_str()));

	//create database class
	fDatabase = new AliStorageDatabase();
	TThread::UnLock();

	//check current storage size
	fCurrentStorageSize = GetSizeOfAllChunks();

	//create two-way commynication thread
	fCommunicationThread = new TThread("fCommunicationThread",
                                       Dispatch,(void*)this);
	fCommunicationThread->Run();
}

AliStorageClientThread::~AliStorageClientThread()
{
	cout<<"CLIENT -- AliStorageClientThread destructor called";
	if(fCurrentFile)
	{
		fCurrentFile->Close();
		delete fCurrentFile;
	}
	if(fCommunicationThread){delete fCommunicationThread;}
	if(fDatabase){delete fDatabase;}
	cout<<" --- OK"<<endl;
}

void AliStorageClientThread::CommunicationHandle()
{
	AliStorageEventManager *eventManager = AliStorageEventManager::GetEventManagerInstance();
	storageSockets socket = CLIENT_COMMUNICATION_REP;
	eventManager->CreateSocket(socket);
	
	struct clientRequestStruct *request;
	struct clientRequestStruct *response = new struct clientRequestStruct;

	cout<<"CLIENT -- Communication stated"<<endl;
	
	while(!gClientQuit)
	{
		request = eventManager->GetClientStruct(socket);
		switch(request->messageType)
		{
		case REQUEST_CONNECTION:
			eventManager->Send((long)fConnectionStatus,socket);
			break;
		case REQUEST_RECEIVING:
			eventManager->Send((long)fReceivingStatus,socket);
			break;
		case REQUEST_SAVING:
			eventManager->Send((long)fSavingStatus,socket);
			break;
		case REQUEST_CURRENT_SIZE:
			eventManager->Send((long)fCurrentStorageSize,socket);
			break;
		case REQUEST_GET_PARAMS:
			response->maxStorageSize = fMaximumStorageSize;
			response->maxOccupation = fStorageOccupationLevel;
			response->removeEvents = fRemoveEventsPercentage;
			response->eventsInChunk = fNumberOfEventsInFile;

			eventManager->Send(response,socket);
			break;
		case REQUEST_SET_PARAMS:
			SetStorageParams(request->maxStorageSize,
					 request->maxOccupation,
					 request->removeEvents,
					 request->eventsInChunk);

			fMaximumStorageSize = request->maxStorageSize;
			fStorageOccupationLevel = request->maxOccupation;
			fRemoveEventsPercentage = request->removeEvents;
			fNumberOfEventsInFile = request->eventsInChunk;

			eventManager->Send(true,socket);
			break;
		default:break;
		}
		delete request;
	}
}

void AliStorageClientThread::SetStorageParams(int maxStorageSize,int maxOccupation,int removeEvents,int eventsInChunk)
{
	cout<<maxStorageSize<<endl<<maxOccupation<<endl<<removeEvents<<endl<<eventsInChunk<<endl;


	TThread::Lock();
	ifstream configFile (GetConfigFilePath());
	ofstream tmpFile("tmpFile.bla");
	
	if (configFile.is_open())
	{
		string line;
		string tmpLine;
		int from,to;
		while(configFile.good())
		{
			getline(configFile,line);
			from = line.find("\"")+1;
			to = line.find_last_of("\"");
			tmpLine = line;
			if(line.find("MAX_SIZE=")==0)
			{
				tmpLine = Form("MAX_SIZE=\"%d\"",maxStorageSize);
			}
			else if(line.find("MAX_OCCUPATION=")==0)
			{
				tmpLine = Form("MAX_OCCUPATION=\"%d\"",maxOccupation);
			}
			else if(line.find("REMOVE_PERCENT=")==0)
			{
				tmpLine = Form("REMOVE_PERCENT=\"%d\"",removeEvents);
			}
			else if(line.find("EVENTS_IN_FILE=")==0)
			{
				tmpLine = Form("EVENTS_IN_FILE=\"%d\"",eventsInChunk);
			}
			tmpLine += "\n";
			tmpFile << tmpLine;
		}
		if(configFile.eof())
		{
			configFile.clear();
		}
		configFile.close();
		tmpFile.close();
		rename("tmpFile.bla",GetConfigFilePath());
	}
	else
	{
		cout<<"CLIENT -- Unable to open config file"<<endl;
	}
	TThread::UnLock();
}

Long64_t AliStorageClientThread::GetSizeOfAllChunks()
{
	Long64_t totalStorageSize = 0;

	TSystemDirectory dir(fStoragePath.c_str(),fStoragePath.c_str());
	TList *listOfDirectories = dir.GetListOfFiles();

	if (!listOfDirectories)
	{
		cout<<"CLIENT -- Storage directory is empty"<<endl;
		return 0;
	}
	TIter nextDirectory(listOfDirectories);
	TSystemFile *runDirectory;
	string directoryName;
	
	while ((runDirectory=(TSystemFile*)nextDirectory()))
	{
		directoryName=runDirectory->GetName();
		if (runDirectory->IsDirectory() && directoryName.find("run")==0)
		{
			TSystemDirectory dirChunks(Form("%s/%s",fStoragePath.c_str(),directoryName.c_str()),Form("%s/%s",fStoragePath.c_str(),directoryName.c_str()));
			TList *listOfChunks = dirChunks.GetListOfFiles();

			if(listOfChunks)
			{
				TIter nextChunk(listOfChunks);
				TSystemFile *chunk;
				string chunkFileName;

				while((chunk=(TSystemFile*)nextChunk()))
				{
					chunkFileName = chunk->GetName();
					if(!chunk->IsDirectory() && chunkFileName.find("chunk")==0)
					{
						TFile *tmpFile = new TFile(Form("%s/%s/%s",fStoragePath.c_str(),directoryName.c_str(),chunkFileName.c_str()),"read");
						if(tmpFile)
						{
							totalStorageSize+=tmpFile->GetSize();
							tmpFile->Close();
							delete tmpFile;
						}
					}
				}
				if(chunk){delete chunk;}
			}
			if(listOfChunks){delete listOfChunks;}
		}
	}

	//tmpFiles.clear();
	if(listOfDirectories){delete listOfDirectories;}
	if(runDirectory){delete runDirectory;}
	
	printf("CLIENT -- Total storage size:%lld\t(%.2f MB)\n",totalStorageSize,(float)totalStorageSize/(1000.*1000.));

	return totalStorageSize;
}

void AliStorageClientThread::CollectData()
{
	AliStorageEventManager *eventManager = AliStorageEventManager::GetEventManagerInstance();
	if(eventManager->CreateSocket(EVENTS_SERVER_SUB))
	{
		fConnectionStatus=STATUS_OK;
	}
	else
	{
		fConnectionStatus=STATUS_ERROR;
	}
	
	int chunkNumber=0;
	int previousChunkNumber=-1;
	int eventsInChunk=0;
	int previousRunNumber=-1;
	AliESDEvent *event = NULL;
//        TTree *tree = NULL;
        
	while(!gClientQuit)
	{		
          event = eventManager->GetEvent(EVENTS_SERVER_SUB);

		if(event)
		{
	       		fReceivingStatus=STATUS_OK;
			
			if(event->GetRunNumber() != previousRunNumber)//when new run starts
			{
				cout<<"CLIENT -- new run started"<<endl;
				previousRunNumber = event->GetRunNumber();
				gSystem->Exec(Form("mkdir -p %s/run%d",fStoragePath.c_str(),event->GetRunNumber()));
				chunkNumber=0;
				eventsInChunk=0;
				
				TSystemDirectory dir(Form("%s/run%d",fStoragePath.c_str(),event->GetRunNumber()),
						     Form("%s/run%d",fStoragePath.c_str(),event->GetRunNumber()));
				TList *files = dir.GetListOfFiles();	
				if (files)
				{
					TSystemFile *file;
					string fname;
					TIter next(files);
					
					while ((file=(TSystemFile*)next()))
					{
						fname = file->GetName();
					
						if (!file->IsDirectory())
						{
							int from = fname.find("chunk")+5;
							int to = fname.find(".root");

							int maxChunkNumber = atoi(fname.substr(from,to-from).c_str());

							if(maxChunkNumber > chunkNumber)
							{
								chunkNumber = maxChunkNumber;
							}
						}
					}
					chunkNumber++;
				}
			}

			cout<<"CLIENT -- Received data. Event:"<<event->GetEventNumberInFile()<<"\trun:"<<event->GetRunNumber()<<endl;
			
			if(chunkNumber != previousChunkNumber)//when new chunk needs to be created
			{
				if(fCurrentFile)
				{
					fCurrentFile->Close();
					delete fCurrentFile;
					fCurrentFile=0;
				}
				fCurrentStorageSize=GetSizeOfAllChunks();
				CheckCurrentStorageSize();
				
				fCurrentFile = new TFile(Form("%s/run%d/chunk%d.root", fStoragePath.c_str(),event->GetRunNumber(),chunkNumber),"recreate");

				previousChunkNumber = chunkNumber;
			}
			
			if(0 != fCurrentFile->WriteObject(event,Form("event%d",event->GetEventNumberInFile())))//if event was written to file
			{
				fDatabase->InsertEvent(event->GetRunNumber(),
					      event->GetEventNumberInFile(),
					      (char*)event->GetBeamType(),
						       event->GetMultiplicity()->GetNumberOfTracklets(),Form("%s/run%d/chunk%d.root",fStoragePath.c_str(),event->GetRunNumber(),chunkNumber));
				
				eventsInChunk++;
				
				if(eventsInChunk == fNumberOfEventsInFile)//if max events number in file was reached
				{
					chunkNumber++;
					eventsInChunk=0;
				}
				
				if(fSavingStatus!=STATUS_OK)
				{
					fSavingStatus=STATUS_OK;
				}
			}
			else if(fSavingStatus!=STATUS_ERROR)
			{
				fSavingStatus=STATUS_ERROR;
			}
			delete event;event=0;
                        //delete tree;
		}
		else if(fReceivingStatus!=STATUS_ERROR)
		{
			cout<<"CLIENT -- ERROR -- NO DATA!"<<endl;
			fReceivingStatus=STATUS_ERROR;
		}
	}
	if(event){delete event;}
}


void AliStorageClientThread::CheckCurrentStorageSize()
{
	if(fCurrentStorageSize >  (float)fStorageOccupationLevel/100. * fMaximumStorageSize)
	{
		while(GetSizeOfAllChunks() > (float)fRemoveEventsPercentage/100. * fMaximumStorageSize)
		{
			struct eventStruct oldestEvent = fDatabase->GetOldestEvent();
			string oldestEventPath = fDatabase->GetFilePath(oldestEvent);
			//remove oldest event
			cout<<"CLIENT -- Removing old events:"<<oldestEventPath<<endl;
			gSystem->Exec(Form("rm -f %s",oldestEventPath.c_str()));
			fDatabase->RemoveEvent(fDatabase->GetOldestEvent());
		}
	}
}
Commit	Line	Data
5eb34a26	1	#include "AliStorageClientThread.h"
	2	#include "AliMultiplicity.h"
	3	#include "AliStorageTypes.h"
5eb34a26	4	#include "AliStorageEventManager.h"
5eb34a26	5
5eb34a26	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;
5eb34a26	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),
5eb34a26	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();
164d3d29	43	ifstream configFile (GetConfigFilePath());
5eb34a26	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	}
5eb34a26	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",
164d3d29	96	Dispatch,(void*)this);
5eb34a26	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;}
a410aca4	109	if(fDatabase){delete fDatabase;}
5eb34a26	110	cout<<" --- OK"<<endl;
	111	}
	112
164d3d29	113	void AliStorageClientThread::CommunicationHandle()
5eb34a26	114	{
164d3d29	115	AliStorageEventManager *eventManager = AliStorageEventManager::GetEventManagerInstance();
	116	storageSockets socket = CLIENT_COMMUNICATION_REP;
	117	eventManager->CreateSocket(socket);
5eb34a26	118
164d3d29	119	struct clientRequestStruct *request;
5eb34a26	120	struct clientRequestStruct *response = new struct clientRequestStruct;
	121
	122	cout<<"CLIENT -- Communication stated"<<endl;
	123
	124	while(!gClientQuit)
	125	{
164d3d29	126	request = eventManager->GetClientStruct(socket);
5eb34a26	127	switch(request->messageType)
	128	{
	129	case REQUEST_CONNECTION:
164d3d29	130	eventManager->Send((long)fConnectionStatus,socket);
5eb34a26	131	break;
5eb34a26	132	case REQUEST_RECEIVING:
164d3d29	133	eventManager->Send((long)fReceivingStatus,socket);
5eb34a26	134	break;
5eb34a26	135	case REQUEST_SAVING:
164d3d29	136	eventManager->Send((long)fSavingStatus,socket);
5eb34a26	137	break;
5eb34a26	138	case REQUEST_CURRENT_SIZE:
164d3d29	139	eventManager->Send((long)fCurrentStorageSize,socket);
5eb34a26	140	break;
5eb34a26	141	case REQUEST_GET_PARAMS:
164d3d29	142	response->maxStorageSize = fMaximumStorageSize;
	143	response->maxOccupation = fStorageOccupationLevel;
	144	response->removeEvents = fRemoveEventsPercentage;
	145	response->eventsInChunk = fNumberOfEventsInFile;
5eb34a26	146
	147	eventManager->Send(response,socket);
	148	break;
	149	case REQUEST_SET_PARAMS:
164d3d29	150	SetStorageParams(request->maxStorageSize,
	151	request->maxOccupation,
	152	request->removeEvents,
	153	request->eventsInChunk);
5eb34a26	154
164d3d29	155	fMaximumStorageSize = request->maxStorageSize;
	156	fStorageOccupationLevel = request->maxOccupation;
	157	fRemoveEventsPercentage = request->removeEvents;
	158	fNumberOfEventsInFile = request->eventsInChunk;
5eb34a26	159
	160	eventManager->Send(true,socket);
	161	break;
	162	default:break;
	163	}
164d3d29	164	delete request;
5eb34a26	165	}
5eb34a26	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();
164d3d29	174	ifstream configFile (GetConfigFilePath());
5eb34a26	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();
164d3d29	213	rename("tmpFile.bla",GetConfigFilePath());
5eb34a26	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");
164d3d29	258	if(tmpFile)
	259	{
	260	totalStorageSize+=tmpFile->GetSize();
	261	tmpFile->Close();
	262	delete tmpFile;
	263	}
5eb34a26	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	{
164d3d29	283	AliStorageEventManager *eventManager = AliStorageEventManager::GetEventManagerInstance();
164d3d29	284	if(eventManager->CreateSocket(EVENTS_SERVER_SUB))
5eb34a26	285	{
164d3d29	286	fConnectionStatus=STATUS_OK;
5eb34a26	287	}
164d3d29	288	else
5eb34a26	289	{
164d3d29	290	fConnectionStatus=STATUS_ERROR;
5eb34a26	291	}
5eb34a26	292
	293	int chunkNumber=0;
	294	int previousChunkNumber=-1;
	295	int eventsInChunk=0;
	296	int previousRunNumber=-1;
	297	AliESDEvent *event = NULL;
164d3d29	298	// TTree *tree = NULL;
164d3d29	299
5eb34a26	300	while(!gClientQuit)
5eb34a26	301	{
164d3d29	302	event = eventManager->GetEvent(EVENTS_SERVER_SUB);
5eb34a26	303
	304	if(event)
	305	{
	306	fReceivingStatus=STATUS_OK;
	307
	308	if(event->GetRunNumber() != previousRunNumber)//when new run starts
	309	{
	310	cout<<"CLIENT -- new run started"<<endl;
	311	previousRunNumber = event->GetRunNumber();
	312	gSystem->Exec(Form("mkdir -p %s/run%d",fStoragePath.c_str(),event->GetRunNumber()));
	313	chunkNumber=0;
	314	eventsInChunk=0;
	315
	316	TSystemDirectory dir(Form("%s/run%d",fStoragePath.c_str(),event->GetRunNumber()),
	317	Form("%s/run%d",fStoragePath.c_str(),event->GetRunNumber()));
	318	TList *files = dir.GetListOfFiles();
	319	if (files)
	320	{
	321	TSystemFile *file;
	322	string fname;
	323	TIter next(files);
	324
	325	while ((file=(TSystemFile*)next()))
	326	{
	327	fname = file->GetName();
	328
	329	if (!file->IsDirectory())
	330	{
	331	int from = fname.find("chunk")+5;
	332	int to = fname.find(".root");
	333
	334	int maxChunkNumber = atoi(fname.substr(from,to-from).c_str());
	335
	336	if(maxChunkNumber > chunkNumber)
	337	{
	338	chunkNumber = maxChunkNumber;
	339	}
	340	}
	341	}
	342	chunkNumber++;
	343	}
	344	}
	345
	346	cout<<"CLIENT -- Received data. Event:"<<event->GetEventNumberInFile()<<"\trun:"<<event->GetRunNumber()<<endl;
	347
	348	if(chunkNumber != previousChunkNumber)//when new chunk needs to be created
	349	{
	350	if(fCurrentFile)
	351	{
	352	fCurrentFile->Close();
	353	delete fCurrentFile;
	354	fCurrentFile=0;
	355	}
	356	fCurrentStorageSize=GetSizeOfAllChunks();
	357	CheckCurrentStorageSize();
	358
	359	fCurrentFile = new TFile(Form("%s/run%d/chunk%d.root", fStoragePath.c_str(),event->GetRunNumber(),chunkNumber),"recreate");
	360
	361	previousChunkNumber = chunkNumber;
	362	}
	363
	364	if(0 != fCurrentFile->WriteObject(event,Form("event%d",event->GetEventNumberInFile())))//if event was written to file
	365	{
	366	fDatabase->InsertEvent(event->GetRunNumber(),
367	event->GetEventNumberInFile(),
368	(char*)event->GetBeamType(),
369	event->GetMultiplicity()->GetNumberOfTracklets(),Form("%s/run%d/chunk%d.root",fStoragePath.c_str(),event->GetRunNumber(),chunkNumber));
370
371	eventsInChunk++;
372
373	if(eventsInChunk == fNumberOfEventsInFile)//if max events number in file was reached
374	{
375	chunkNumber++;
376	eventsInChunk=0;
377	}
378
379	if(fSavingStatus!=STATUS_OK)
380	{
381	fSavingStatus=STATUS_OK;
382	}
383	}
384	else if(fSavingStatus!=STATUS_ERROR)
385	{
386	fSavingStatus=STATUS_ERROR;
387	}
388	delete event;event=0;
164d3d29	389	//delete tree;
5eb34a26	390	}
	391	else if(fReceivingStatus!=STATUS_ERROR)
	392	{
	393	cout<<"CLIENT -- ERROR -- NO DATA!"<<endl;
	394	fReceivingStatus=STATUS_ERROR;
	395	}
	396	}
	397	if(event){delete event;}
	398	}
	399
	400
	401	void AliStorageClientThread::CheckCurrentStorageSize()
	402	{
	403	if(fCurrentStorageSize > (float)fStorageOccupationLevel/100. * fMaximumStorageSize)
	404	{
	405	while(GetSizeOfAllChunks() > (float)fRemoveEventsPercentage/100. * fMaximumStorageSize)
	406	{
	407	struct eventStruct oldestEvent = fDatabase->GetOldestEvent();
	408	string oldestEventPath = fDatabase->GetFilePath(oldestEvent);
	409	//remove oldest event
	410	cout<<"CLIENT -- Removing old events:"<<oldestEventPath<<endl;
	411	gSystem->Exec(Form("rm -f %s",oldestEventPath.c_str()));
	412	fDatabase->RemoveEvent(fDatabase->GetOldestEvent());
	413	}
	414	}
	415	}