/*
    Copyright (C) 2012 Bartosz Broda, Jacek Bochenek, Łukasz Biłeńkij, 
    Damian Jaworski, Wojciech Mazur, Dominik Piasecki, Maciej Piasecki, 
    Radosław Ramocki
    
    This program is free software; you can redistribute it and/or modify it
under the terms of the GNU Lesser General Public License as published by the Free
Software Foundation; either version 3 of the License, or (at your option)
any later version.

    This program is distributed in the hope that it will be useful, but
WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
or FITNESS FOR A PARTICULAR PURPOSE. 

    See the LICENSE and COPYING files for more details.
*/

#include "parallelArchitect.h"


void parallelMasterDict(MPI_Status* stat)
{
	int numOfNodes;
	MPI_Comm_size(MPI_COMM_WORLD,&numOfNodes);
	if(numOfNodes==1)
	{
		onlyMasterDict();
		return;
	}
	me_architect::Config *conf = me_architect::Config::GetInstance();
	boost::filesystem::create_directories( conf->m_sWorkdir);
	udict_map base_udict; // flex -> ( id -> count)
	for(std::set<Corpus::TFlexClass>::iterator iter = conf->flexes.begin(); iter != conf->flexes.end(); iter++)
		base_udict[*iter];// = std::map<int, size_t>();
	Corpus::CWordDictionary *dict = Corpus::CWordDictionary::GetInstance();
	int totalNumOfTasks = conf->m_sCorpusFiles.size();
	int nodesFinished = 0;
	int taskIterator = 0;
	int probeFlag = 0;
	int sourceNode;
	int message;

	while(nodesFinished<numOfNodes-1)
	{
		MPI_Iprobe(MPI_ANY_SOURCE,0,MPI_COMM_WORLD,&probeFlag,stat);
		while(probeFlag==0)
		{
			sleep(1);
			MPI_Iprobe(MPI_ANY_SOURCE,0,MPI_COMM_WORLD,&probeFlag,stat);
		}

		MPI_Recv(&message,1,MPI_INT,MPI_ANY_SOURCE,0,MPI_COMM_WORLD,stat);
		sourceNode = stat->MPI_SOURCE;

		if(message == GIVE_WORK)
		{
			if(taskIterator==totalNumOfTasks)
			{
				message = EXIT_CODE;
				MPI_Send(&message,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD);
			}
			else
			{
				message=taskIterator;
				MPI_Send(&message,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD);
				taskIterator++;
			}
		}
		if(message == SENDING_DICT)
		{
			int numOfFlexes;
			MPI_Recv(&numOfFlexes,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);
			int lexemsPerFlex[numOfFlexes];
			MPI_Recv(lexemsPerFlex,numOfFlexes,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);
			int flexClassLength[numOfFlexes];
			MPI_Recv(flexClassLength,numOfFlexes,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);
			int wordLengthTableLength;
			MPI_Recv(&wordLengthTableLength,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);
			int *wordLengthTable = new int[wordLengthTableLength];
			int *wordCountTable = new int[wordLengthTableLength];
			MPI_Recv(wordLengthTable,wordLengthTableLength,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);
			MPI_Recv(wordCountTable,wordLengthTableLength,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);
			int flexCharTableLength;
			MPI_Recv(&flexCharTableLength,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);
			char flexCharTable[flexCharTableLength];
			MPI_Recv(flexCharTable,flexCharTableLength,MPI_CHAR,sourceNode,0,MPI_COMM_WORLD,stat);
			int wordCharTableLength;
			MPI_Recv(&wordCharTableLength,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);
			char *wordCharTable = new char[wordCharTableLength];
			MPI_Recv(wordCharTable,wordCharTableLength,MPI_CHAR,sourceNode,0,MPI_COMM_WORLD,stat);

			std::cout<<"Received dict from node "<<sourceNode<<", processing...\n";

			int flexCharTablePosition=0;
			int wordCharTablePosition=0;
			int wordCounter=0;

			for(int i=0;i<numOfFlexes;i++)
			{
				char charFlex[flexClassLength[i]+1];
				strncpy(charFlex,&(flexCharTable[flexCharTablePosition]),flexClassLength[i]);
				flexCharTablePosition+=flexClassLength[i];
				charFlex[flexClassLength[i]]='\0';
				std::string flexString(charFlex);
				Corpus::TFlexClass flexID = Corpus::CTag::GetFlexClass(flexString);
				for(int j=0;j<lexemsPerFlex[i];j++)
				{
					char currentWordChar[wordLengthTable[wordCounter]+1];
					strncpy(currentWordChar,&(wordCharTable[wordCharTablePosition]),wordLengthTable[wordCounter]);
					wordCharTablePosition+=wordLengthTable[wordCounter];
					currentWordChar[wordLengthTable[wordCounter]]='\0';

					std::string currentWordString(currentWordChar);
					int currentWordID = dict->Add(currentWordString);
					udict_inner_map::iterator find = base_udict[flexID].find(currentWordID);



					if(find==base_udict[flexID].end())
					{
						base_udict[flexID].insert( std::make_pair( currentWordID, wordCountTable[wordCounter] ) );
					}
					else
					{
						find->second+=wordCountTable[wordCounter];
					}

					wordCounter++;
				}

			}
			delete [] wordLengthTable;
			delete [] wordCountTable;
			delete [] wordCharTable;
			std::cout<<"Processed dictionary from node "<<sourceNode<<std::endl;
			nodesFinished++;


		}
	}


	std::cout << "\nWriting data..." << std::endl;



	write_bases(*conf, base_udict);

}
std::set<std::string> loadStopList()
{
	me_architect::Config* conf = me_architect::Config::GetInstance();
	std::string line;
	std::set<std::string> stopList;
	for(int i=0;i<(int)conf->stoplistFiles.size();i++)
	{
		std::ifstream stopListFile(conf->stoplistFiles[i].c_str());
		if(!stopListFile)
			throw std::runtime_error("Specified stoplist file not found!\n");
		while((std::getline(stopListFile,line)))
			stopList.insert(line);
		stopListFile.close();
	}
	return stopList;
}

void parallelSlaveDict(MPI_Status* stat)
{
	me_architect::Config* conf = me_architect::Config::GetInstance();
	udict_map base_udict; // flex -> ( id -> count)
	for(std::set<Corpus::TFlexClass>::iterator iter = conf->flexes.begin(); iter != conf->flexes.end(); iter++)
		base_udict[*iter];// = std::map<int, size_t>();
	bool exitFlag = false;
	int myRank;
	MPI_Comm_rank(MPI_COMM_WORLD,&myRank);

	int message;

	std::set<std::string> stopList = loadStopList();



	while(!exitFlag)
	{
		int message = GIVE_WORK;
		MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);

		MPI_Recv(&message,1,MPI_INT,0,0,MPI_COMM_WORLD,stat);
		if(message==EXIT_CODE)
			exitFlag = true;
		else
		{
			singleCorpusDict(message,base_udict);
			std::cout<<"Processed corpus "<<message<<" on node "<<myRank<<std::endl;
		}

	}
	//Preparing dict to be send

	Corpus::CWordDictionary *dict = Corpus::CWordDictionary::GetInstance();
	int flexClassCharTableLength=0;
	int numOfFlexes = base_udict.size();
	int lexemsPerFlex[numOfFlexes];
	int flexClassLength[numOfFlexes];

	int wordCharTableLength=0;
	int wordLengthTableLength=0;
	int counter=0;
	for(udict_map::iterator iter = base_udict.begin();iter!=base_udict.end();++iter)
	{
		flexClassCharTableLength+=strlen(Corpus::CTag::GetFlexClassString( iter->first ));


		udict_inner_map &inner = iter->second;

		flexClassLength[counter]=strlen(Corpus::CTag::GetFlexClassString( iter->first ));

		lexemsPerFlex[counter]=0;
		for(udict_inner_map::iterator inner_iter = inner.begin(); inner_iter != inner.end(); ++inner_iter)
		{
			if((stopList.find(dict->GetAt(inner_iter->first))==stopList.end()))
			{
				wordCharTableLength+=dict->GetAt( inner_iter->first).length();
				wordLengthTableLength++;
				lexemsPerFlex[counter]++;
			}
		}
		counter++;
	}
	char *wordCharTable = new char[wordCharTableLength];
	char flexClassCharTable[flexClassCharTableLength];
	int * wordLengthTable = new int[wordLengthTableLength];
	int * wordCountTable = new int[wordLengthTableLength];

	int flexTablePosition=0;
	int wordCharTablePosition=0;
	int wordLengthTablePosition=0;

	for(udict_map::iterator iter = base_udict.begin();iter!=base_udict.end();++iter)
	{
		memcpy(&(flexClassCharTable[flexTablePosition]),Corpus::CTag::GetFlexClassString( iter->first ),strlen((Corpus::CTag::GetFlexClassString( iter->first ))));
		flexTablePosition+=strlen(Corpus::CTag::GetFlexClassString( iter->first ));

		udict_inner_map &inner = iter->second;
		for(udict_inner_map::iterator inner_iter = inner.begin(); inner_iter != inner.end(); ++inner_iter)
		{
			if((stopList.find(dict->GetAt(inner_iter->first))==stopList.end()))
			{
				memcpy(&(wordCharTable[wordCharTablePosition]),dict->GetAt( inner_iter->first).c_str(),dict->GetAt( inner_iter->first).length());
				wordCharTablePosition+=dict->GetAt( inner_iter->first).length();
				wordLengthTable[wordLengthTablePosition]=dict->GetAt( inner_iter->first).length();
				wordCountTable[wordLengthTablePosition] = inner_iter->second;
				wordLengthTablePosition++;
			}
		}
	}

	std::cout<<"Prepared dictionary on node "<<myRank<<std::endl;
	message=SENDING_DICT;
	MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);
	MPI_Send(&numOfFlexes,1,MPI_INT,0,0,MPI_COMM_WORLD);
	MPI_Send(lexemsPerFlex,numOfFlexes,MPI_INT,0,0,MPI_COMM_WORLD);
	MPI_Send(flexClassLength,numOfFlexes,MPI_INT,0,0,MPI_COMM_WORLD);
	MPI_Send(&wordLengthTableLength,1,MPI_INT,0,0,MPI_COMM_WORLD);
	MPI_Send(wordLengthTable,wordLengthTableLength,MPI_INT,0,0,MPI_COMM_WORLD);
	MPI_Send(wordCountTable,wordLengthTableLength,MPI_INT,0,0,MPI_COMM_WORLD);
	MPI_Send(&flexClassCharTableLength,1,MPI_INT,0,0,MPI_COMM_WORLD);
	MPI_Send(flexClassCharTable,flexClassCharTableLength,MPI_CHAR,0,0,MPI_COMM_WORLD);
	MPI_Send(&wordCharTableLength,1,MPI_INT,0,0,MPI_COMM_WORLD);
	MPI_Send(wordCharTable,wordCharTableLength,MPI_CHAR,0,0,MPI_COMM_WORLD);
	delete [] wordCharTable;
	delete [] wordCountTable;
	delete [] wordLengthTable;
	std::cout<<"Sent dictionary from node "<<myRank<<std::endl;
	for(udict_map::iterator iter=base_udict.begin();iter!=base_udict.end();++iter)
		iter->second.clear();
	base_udict.clear();

}

void parallelArchitectMaster(int ac, char *av[],MPI_Status*stat)
{

	int numOfNodes;
	MPI_Comm_size(MPI_COMM_WORLD,&numOfNodes);

	std::vector<Triplet_OperatorFlexLex*> triplets= me_architect::Config::InitalizeForMasterNode(ac,av);

	int numOfOperators = triplets[triplets.size()-1]->nextIndex;
	int numOfOperatorsPerNode = me_architect::Config::GetInstance()->GetOperatorsPerNode();
	std::cout<<"Num of operators: "<<numOfOperators<<std::endl;
	

	if(numOfOperatorsPerNode==-1)
		numOfOperatorsPerNode=numOfOperators/(numOfNodes-1) +1; //On default, last node will process couple columns less,

	std::vector<int> taskBoundaries;

	int previousOp = triplets[0]->operatorIndex;
	for(int i=0;i<(int)triplets.size();i++)
	{
		if(previousOp!=triplets[i]->operatorIndex)
		{
			taskBoundaries.push_back(triplets[i-1]->nextIndex);
			previousOp = triplets[i]->operatorIndex;
		}
	}
	taskBoundaries.push_back(triplets[triplets.size()-1]->nextIndex);




	//in order to make it all at once
	std::vector<std::pair<int,int> > tasks;


	int currentOperator = 0;
	for(int i=0;i<(int)taskBoundaries.size();i++)
	{
		while(currentOperator<taskBoundaries[i])
		{
			if(currentOperator+numOfOperatorsPerNode >=taskBoundaries[i])
			{
				tasks.push_back(std::pair<int,int>(currentOperator,taskBoundaries[i]));
				currentOperator=taskBoundaries[i];
			}
			else
			{
				tasks.push_back(std::pair<int,int>(currentOperator,currentOperator+numOfOperatorsPerNode));
				currentOperator+=numOfOperatorsPerNode;
			}
		}
	}

	std::cout<<"Read "<<numOfOperators<<" operators on master, processing "<<numOfOperatorsPerNode<<" on each node.\n";

	int numOfTasksDone = 0;

	int probeFlag=0;

	int corpusSize=-1;




	int message;
	int sourceNode;
	int *numOfValuesInColumns;
	int *tableOfIndexes;
	double *tableOfValues;
	int columnRangeMin,columnRangeMax;
	int numOfColumns;
	int numOfNonZero;
	int counter;

	std::vector<std::pair<int,int> >::iterator tasksIterator=tasks.begin();

	int numOfRows = me_architect::Config::GetInstance()->rowLU.size();
	Matrices::CCSMatrixPtr ccsMatrix( new Matrices::CCSMatrix("ccs",numOfRows,numOfOperators));

	int numOfLUs;
	int *tableLULength;
	int *tableTF;
	int lengthCharTable;
	char *charTable;
	std::map<std::string,size_t> wordTF;
	char*buffer;


	int shuffleTable[numOfOperators];
	std::vector<int> shuffleVector;
	for(int i=0;i<numOfOperators;i++)
		shuffleVector.push_back(i);


	std::vector<int>::iterator startIter,endIter;
	startIter = shuffleVector.begin();
	for(int i=0;i<(int)taskBoundaries.size();i++)
	{
		//		endIter = startIter;
		endIter=shuffleVector.begin()+taskBoundaries[i];
		std::random_shuffle(startIter,endIter);
		startIter = endIter;
	}



	for(int i=0;i<numOfOperators;i++)
	{
		shuffleTable[i]=shuffleVector[i];

	}


	int timeCounter=0;
	while(numOfTasksDone<(int)tasks.size())
	{
		MPI_Iprobe(MPI_ANY_SOURCE,0,MPI_COMM_WORLD,&probeFlag,stat);
		while(probeFlag==0)
		{
			sleep(1);
			timeCounter++;
			if(timeCounter==50)
			{
				std::cout<<"Master: Finished "<<numOfTasksDone<<" out of "<<tasks.size()<<std::endl;
				FILE* popen_stream;
				popen_stream = popen("free -m | head -n 3 | tail -n 1 | awk -F ' '  '{print $4}'","r");
				char * ram = new char[30];
				fgets(ram,30, popen_stream);
				std::string ram_str(ram);
				boost::trim(ram_str);
				int val = boost::lexical_cast<int>(ram_str);
				if (val<128)
				{
					std::cout<<"\033[31;1m!!!!!!WARNING!!!!!! ONLY "<< val << "MB OF FREE RAM LEFT ON MASTER NODE!!!\033[0m\n";
				}
				else
				{
					if (val<512)
					{
						std::cout<<"Warning: only "<< val << " MB of free RAM left on master node.\n";
					}
				}
				pclose(popen_stream);
				delete [] ram;
				timeCounter=0;
			}
			MPI_Iprobe(MPI_ANY_SOURCE,0,MPI_COMM_WORLD,&probeFlag,stat);
		}


		MPI_Recv(&message,1,MPI_INT,MPI_ANY_SOURCE,0,MPI_COMM_WORLD,stat);
		sourceNode = stat->MPI_SOURCE;

		if(message==GIVE_SHUFFLE_TABLE)
		{
			message=numOfOperators;
			MPI_Send(&message,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD);

			MPI_Send(shuffleTable,numOfOperators,MPI_INT,sourceNode,0,MPI_COMM_WORLD);
		}
		else if(message==GIVE_WORK)
		{
			if(tasksIterator==tasks.end())
			{
				message=EXIT_CODE;
				MPI_Send(&message,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD);
			}
			else
			{
				message=ALL_OK;
				MPI_Send(&message,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD);

				message=(*tasksIterator).first;
				MPI_Send(&message,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD);

				message=(*tasksIterator).second;
				MPI_Send(&message,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD);

				tasksIterator++;
			}
		}
		if(message==SENDING_RESULTS)
		{
			MPI_Recv(&columnRangeMin,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);

			MPI_Recv(&columnRangeMax,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);

			numOfColumns=columnRangeMax - columnRangeMin;
			numOfValuesInColumns = new int[numOfColumns];

			MPI_Recv(numOfValuesInColumns,numOfColumns,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);

			MPI_Recv(&numOfNonZero,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);

			tableOfIndexes = new int[numOfNonZero];
			tableOfValues = new double[numOfNonZero];

			MPI_Recv(tableOfIndexes,numOfNonZero,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);


			MPI_Recv(tableOfValues,numOfNonZero,MPI_DOUBLE,sourceNode,0,MPI_COMM_WORLD,stat);

			//Filling matrices
			counter=0;
			for(int i=columnRangeMin;i<columnRangeMax;i++)
			{


				for(int j=0;j<numOfValuesInColumns[i-columnRangeMin];j++)
				{

					ccsMatrix->SetCell(tableOfValues[counter],tableOfIndexes[counter],shuffleTable[i]);
					counter++;
				}

			}
			if(counter!=numOfNonZero)
				throw std::runtime_error("Error - counter!=numOfNonZero\n");
			delete [] numOfValuesInColumns;
			delete [] tableOfIndexes;
			delete [] tableOfValues;
			std::cout<<"Received part of matrix from node "<<sourceNode<<std::endl;

		}
		else if(message==SENDING_CORPUS_SIZE)
		{
			MPI_Recv(&corpusSize,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);
		}
		else if(message==SENDING_DICT)
		{
			message=ALL_OK;
			MPI_Send(&message,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD);


			MPI_Recv(&message,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);
			numOfLUs=message;



			tableLULength = new int[numOfLUs];
			tableTF = new int[numOfLUs];

			MPI_Recv(tableLULength,numOfLUs,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);

			MPI_Recv(tableTF,numOfLUs,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);

			MPI_Recv(&lengthCharTable,1,MPI_INT,sourceNode,0,MPI_COMM_WORLD,stat);

			charTable = new char[lengthCharTable];
			MPI_Recv(charTable,lengthCharTable,MPI_CHAR,sourceNode,0,MPI_COMM_WORLD,stat);

			counter=0;
			for(int i=0;i<numOfLUs;i++)
			{
				buffer = new char[tableLULength[i]+1];
				strncpy(buffer,&(charTable[counter]),tableLULength[i]);
				counter+=tableLULength[i];
				buffer[tableLULength[i]]='\0';
				wordTF[std::string(buffer)]=tableTF[i];

				delete [] buffer;
			}

			if(counter!=lengthCharTable)
				throw std::runtime_error("Error: counter!=lengthCharTable");

			delete [] charTable;
			delete [] tableLULength;
			delete [] tableTF;
			std::cout<<"Received dictionary from node "<<sourceNode<<"\n";
			numOfTasksDone++;
		}
	}

	message=EXIT_CODE;
	MPI_Request request;
	for(int i=1;i<numOfNodes;i++)
	{
		MPI_Isend(&message,1,MPI_INT,i,0,MPI_COMM_WORLD,&request);
		MPI_Request_free(&request);
	}

	std::vector<std::string> rowLabels;
	std::vector<std::string> colLabels;
	std::vector<LexicalUnits::LexicalUnitPtr> LUVector = me_architect::Config::GetInstance()->m_vLexicalUnits;
	std::vector<size_t> rowLU = me_architect::Config::GetInstance()->rowLU;
	std::vector<size_t> oldColumnLU = me_architect::Config::GetInstance()->columnLU;

	std::vector<size_t> columnLU;
	//Removing empty columns
	std::set<size_t> columnIndicesToRemoveSet;
	for(int i=0;i<(int)ccsMatrix->NumberOfColumns();i++)
		if(ccsMatrix->GetCompressedColumn(i)->GetValues()->size()==0)
		{
		columnIndicesToRemoveSet.insert(i);
	}



	for(int i=0;i<(int)oldColumnLU.size();i++)
	{
		if(columnIndicesToRemoveSet.find(i)==columnIndicesToRemoveSet.end())
			columnLU.push_back(oldColumnLU[i]);
	}
	Matrices::CCSMatrixPtr newCCSMatrixPtr = Matrices::FilterCCS(ccsMatrix,columnIndicesToRemoveSet);



	Corpus::CWordDictionary *dict = Corpus::CWordDictionary::GetInstance();
	int bufferLength;
	for(std::vector<size_t>::iterator iter = rowLU.begin();iter!=rowLU.end();iter++)
	{
		bufferLength=2; // ":", '\0'
		bufferLength+=(dict->GetAt(LUVector[(*iter)]->GetBaseId())).length();
		bufferLength+=strlen(Corpus::CTag::GetFlexClassString(LUVector[(*iter)]->GetFlexClass()));
		buffer = new char[bufferLength];

		strcpy(buffer,Corpus::CTag::GetFlexClassString(LUVector[(*iter)]->GetFlexClass()));
		buffer[strlen(Corpus::CTag::GetFlexClassString(LUVector[(*iter)]->GetFlexClass()))] = ':';
		strcpy(&(buffer[strlen(Corpus::CTag::GetFlexClassString(LUVector[(*iter)]->GetFlexClass()))+1]),
			   dict->GetAt(LUVector[(*iter)]->GetBaseId()).c_str());

		rowLabels.push_back(std::string(buffer));
		delete [] buffer;
	}

	for(std::vector<size_t>::iterator iter = columnLU.begin();iter!=columnLU.end();iter++)
	{
		bufferLength=2; // ":", '\0'
		bufferLength+=(dict->GetAt(LUVector[(*iter)]->GetBaseId())).length();
		bufferLength+=strlen(Corpus::CTag::GetFlexClassString(LUVector[(*iter)]->GetFlexClass()));
		buffer = new char[bufferLength];

		strcpy(buffer,Corpus::CTag::GetFlexClassString(LUVector[(*iter)]->GetFlexClass()));
		buffer[strlen(Corpus::CTag::GetFlexClassString(LUVector[(*iter)]->GetFlexClass()))] = ':';
		strcpy(&(buffer[strlen(Corpus::CTag::GetFlexClassString(LUVector[(*iter)]->GetFlexClass()))+1]),
			   dict->GetAt(LUVector[(*iter)]->GetBaseId()).c_str());
		colLabels.push_back(std::string(buffer));

		delete [] buffer;

	}
	std::cout<<"Created "<< rowLabels.size() <<" row and "<< colLabels.size()<<" column labels on master\n";

	Matrices::FeatureVectorPtr rowFeatures;
	Matrices::FeatureVectorPtr colFeatures;
	Matrices::CRSMatrixPtr crsMatrix(new Matrices::CRSMatrix("crs Matrix",newCCSMatrixPtr->NumberOfRows(),newCCSMatrixPtr->NumberOfColumns()));
	newCCSMatrixPtr->Clone(crsMatrix);

	PrepareFeatures(rowFeatures,crsMatrix,colFeatures,newCCSMatrixPtr,rowLabels,colLabels,wordTF);

	std::cout<<"Features prepared\n";
	me_architect::Config *c = me_architect::Config::GetInstance();

	//Boundaries


	std::map<size_t,std::string> boundaryMap;

	std::vector<std::pair<std::string,size_t> > opCol = c->GetOperators_Columns();
	std::string previousOperator = opCol[0].first;

	//declared earlier
	counter=0;
	for(int i=0;i<(int)opCol.size();i++)
	{
		if(previousOperator.compare(opCol[i].first) != 0)
		{
			if(counter!=0)
			{
                std::string p = boost::filesystem::path(opCol[i-1].first).filename().c_str();
                boundaryMap[counter] = p;
				counter=0;
			}
			previousOperator = opCol[i].first;
		}
		counter+=opCol[i].second;

	}
    if(counter!=0)
    {
        std::string p = boost::filesystem::path(opCol[opCol.size()-1].first).filename().c_str();
        boundaryMap[counter]= p;
    }

	std::cout<<"Boundaries prepared\n";



	Matrices::SuperMatrixPtr supermatrix( new Matrices::SuperMatrix( c->m_sMatrixName, newCCSMatrixPtr,
																	 crsMatrix,rowFeatures, colFeatures, corpusSize,
																	 c->m_sCorpusInfo,boundaryMap) ) ;

	boundaryMap = Matrices::UpdateBoundaries(supermatrix,columnIndicesToRemoveSet);
	supermatrix->SetBoundaries(boundaryMap);
	Matrices::IO::SuperMatrixWriter writer;
	std::cout << "Saving matrix...\n";
	writer.Write(supermatrix, c->m_sWorkdir);

	//udict.txt

	std::ofstream udictFile((c->m_sWorkdir + c->m_sMatrixName +"/udict.txt").c_str());
	if(!udictFile)
		std::cout<<"Cannot open udict.txt"<<std::endl;

	for(std::map<std::string,size_t>::iterator i=wordTF.begin();i!=wordTF.end();i++)
		if((*i).second>0)
			udictFile<<(*i).first<<":= "<<(*i).second<<std::endl;
	udictFile.close();


}


void PrepareFeatures(Matrices::FeatureVectorPtr &m_pRowFeatures,Matrices::CRSMatrixPtr m_pCRSMatrix,
					 Matrices::FeatureVectorPtr &m_pColumnFeatures,Matrices::CCSMatrixPtr m_pCCSMatrix,
					 std::vector<std::string> m_vRowLabels,std::vector<std::string> m_vColumnLabels,
					 std::map<std::string,size_t> m_mWordTF)
{
	using namespace boost::posix_time;
	Matrices::SuperMatrixPtr pMatrix;

	ptime start = microsec_clock::local_time();
	time_duration total_time;
	m_pRowFeatures = Matrices::FeatureVectorPtr( new Matrices::FeatureVector() );
	std::cout<<"Counting row features...\n";
	for( size_t row_no = 0; row_no < m_pCRSMatrix->NumberOfRows(); row_no++)
	{
		Matrices::CompressedVectorPtr v = m_pCRSMatrix->GetCompressedRow(row_no);
		double entropy = Measures::CountVectorEntropy( *v );
		size_t nz = v->GetValues()->size();
		size_t tf = (size_t) Measures::CountVectorSum( *v );
		std::string word = m_vRowLabels[row_no];

		size_t TF = tf;
		std::map<std::string, size_t>::iterator find = m_mWordTF.find( word );
		if( find != m_mWordTF.end() )
			TF = find->second;
		else
		{
			if(tf > 0)
				std::cout << "No TF (row) information for " << word << ", using tf(" << tf << ")." << std::endl;
		}

		m_pRowFeatures->push_back( Matrices::Feature(word, nz, tf, TF, entropy) ) ;

		ptime end = microsec_clock::local_time();
		time_duration len = end-start;
		if(len.total_milliseconds() > 5000)
		{
			total_time += len;
			long total_ms = total_time.total_milliseconds();
			long eta_ms =  (total_ms*(m_pCRSMatrix->NumberOfRows()-row_no-1)) / (row_no+1);
			start = microsec_clock::local_time();
			std::cout << "Counting row features... " << row_no << " / " << m_pCRSMatrix->NumberOfRows()
					<< " ETA: " << to_simple_string( milliseconds(eta_ms) )  << "           \r";
			std::cout.flush();
		}
	}
	std::cout << std::endl;
	// column features...
	std::cout<<"Counting column features...\n";
	start = microsec_clock::local_time();
	time_duration total_time2;
	m_pColumnFeatures = Matrices::FeatureVectorPtr( new Matrices::FeatureVector() );
	for( size_t column_no = 0; column_no < m_pCCSMatrix->NumberOfColumns(); column_no++)
	{
		Matrices::CompressedVectorPtr v = m_pCCSMatrix->GetCompressedColumn(column_no);
		double entropy = Measures::CountVectorEntropy( *v );
		size_t nz = v->GetValues()->size();
		size_t tf = (size_t) Measures::CountVectorSum( *v );
		std::string word = m_vColumnLabels[column_no];
		

		std::map<std::string, size_t>::iterator find = m_mWordTF.find( word );
		size_t TF = tf;
		if( find != m_mWordTF.end() )
			TF = find->second;
		else
			if(tf > 0)
				std::cout << "No TF (col) information for " << word << ", using tf(" << tf << ")." << std::endl;

		m_pColumnFeatures->push_back( Matrices::Feature(word, nz, tf, TF, entropy) ) ;
		ptime end = microsec_clock::local_time();
		time_duration len = end-start;
		if(len.total_milliseconds() > 50)
		{
			total_time += len;
			long total_ms = total_time.total_milliseconds();
			long eta_ms =  (total_ms*(m_pCCSMatrix->NumberOfColumns()-column_no-1)) / (column_no+1);
			start = microsec_clock::local_time();
			std::cout << "Counting column features... " << column_no << " / " << m_pCCSMatrix->NumberOfColumns()
					<< " ETA: " << to_simple_string( milliseconds(eta_ms) )  << "          \r";
			std::cout.flush();
		}
	}
	std::cout << std::endl;



	std::cout<<"All features prepared. \n";

}
void parallelArchitectSlave(int ac, char *av[],MPI_Status* stat)
{
	int columnRangeMin;
	int columnRangeMax;

	int myRank;
	MPI_Comm_rank(MPI_COMM_WORLD,&myRank);
	int message;
	message=GIVE_SHUFFLE_TABLE;
	MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);

	MPI_Recv(&message,1,MPI_INT,0,0,MPI_COMM_WORLD,stat);

	//	std::cout<<"ShuffleTable length: "<<message<<std::endl;
	int shuffleTable[message];

	MPI_Recv(shuffleTable,message,MPI_INT,0,0,MPI_COMM_WORLD,stat);

	bool exitFlag=false;


	while(!exitFlag)
	{
		
		message=GIVE_WORK;
		MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);

		MPI_Recv(&message,1,MPI_INT,0,0,MPI_COMM_WORLD,stat);

		if(message==EXIT_CODE)
			exitFlag=true;
		else if( message==ALL_OK)
		{

			MPI_Recv(&columnRangeMin,1,MPI_INT,0,0,MPI_COMM_WORLD,stat);
			MPI_Recv(&columnRangeMax,1,MPI_INT,0,0,MPI_COMM_WORLD,stat);

			std::cout<<"Received "<<columnRangeMin<<" and "<<columnRangeMax<<" on node "<<myRank<<std::endl;



			std::cout<<"Initializing...\n";
			me_architect::Config::Initalize(ac,av,columnRangeMin,columnRangeMax,shuffleTable);
			me_architect::Architect arch;


			arch.Start();


			//Sending part of matrix to master

			Matrices::CCSMatrixPtr ptr = arch.getMatrix();

			int numOfColumns = columnRangeMax - columnRangeMin;
			int numOfNonZero = 0;

			int *numOfValuesInColumns;
			numOfValuesInColumns = new int[numOfColumns];

			Matrices::STDoubleMapPtr doubleMapPtr;
			for(int i=0;i<numOfColumns;i++)
			{
				doubleMapPtr = ptr->GetCompressedColumn(i)->GetValues();

				numOfValuesInColumns[i]=doubleMapPtr->size();
				numOfNonZero+=doubleMapPtr->size();
			}
			int* tableOfIndexes;
			tableOfIndexes = new int[numOfNonZero];
			double* tableOfValues;
			tableOfValues = new double[numOfNonZero];
			int counter=0;
			for(int i=0;i<numOfColumns;i++)
			{



				doubleMapPtr = ptr->GetCompressedColumn(i)->GetValues();



				for(std::map<size_t,double>::iterator j=doubleMapPtr->begin();j!=doubleMapPtr->end();j++)
				{

					tableOfIndexes[counter]=j->first;
					tableOfValues[counter]=j->second;

					counter++;
				}
			}
			if(counter!=numOfNonZero)
			{
				std::cout<<"Uknown error - counter!=numOfNonZero \n";
				exit(-1);
			}
			message=SENDING_RESULTS;
			MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);


			message=columnRangeMin;
			MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);

			message=columnRangeMax;
			MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);


			MPI_Send(numOfValuesInColumns,numOfColumns,MPI_INT,0,0,MPI_COMM_WORLD);

			MPI_Send(&numOfNonZero,1,MPI_INT,0,0,MPI_COMM_WORLD);

			MPI_Send(tableOfIndexes,numOfNonZero,MPI_INT,0,0,MPI_COMM_WORLD);

			MPI_Send(tableOfValues,numOfNonZero,MPI_DOUBLE,0,0,MPI_COMM_WORLD);

			std::cout<<"Node "<<myRank<<" sent part of matrix to master\n";
			if(myRank==1)
			{
				message=SENDING_CORPUS_SIZE;
				MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);

				message=arch.getCorpusSize();
				MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);
				std::cout<<"Node "<<myRank<<" sent corpus size to master\n";
			}
			//Sending dictionary
			LexicalUnits::UnigramDictionary *udict = LexicalUnits::UnigramDictionary::GetInstance();
			Corpus::CWordDictionary *dict = Corpus::CWordDictionary::GetInstance();
			me_architect::Config *config = me_architect::Config::GetInstance();

			std::vector<LexicalUnits::LexicalUnitPtr> lexUn = config->m_vLexicalUnits;

			int tableTF[lexUn.size()];
			int tableLULength[lexUn.size()];
			counter=0;
			int lengthSingleLU;
			int numOfLexUnits=0;

			for(int i=0;i<(int)lexUn.size();i++)
			{
				if(udict->GetCount(lexUn[i]) >0)
				{
					tableTF[numOfLexUnits] = udict->GetCount(lexUn[i]);

					lengthSingleLU=0;
					lengthSingleLU+=strlen(Corpus::CTag::GetFlexClassString(lexUn[i]->GetFlexClass()));
					lengthSingleLU+=1; // ":"
					lengthSingleLU+=dict->GetAt(lexUn[i]->GetBaseId()).length();
					tableLULength[numOfLexUnits]=lengthSingleLU;


					counter+=lengthSingleLU;
					numOfLexUnits++;
				}
			}

			char *charTable = new char[counter];
			counter=0;


			for(int i=0;i<(int)lexUn.size();i++)
			{
				if(udict->GetCount(lexUn[i]) >0)
				{
				memcpy(&(charTable[counter]),Corpus::CTag::GetFlexClassString(lexUn[i]->GetFlexClass()),strlen(Corpus::CTag::GetFlexClassString(lexUn[i]->GetFlexClass())));
				counter+=strlen(Corpus::CTag::GetFlexClassString(lexUn[i]->GetFlexClass()));
				charTable[counter]=':';
				counter++;
				memcpy(&(charTable[counter]),dict->GetAt(lexUn[i]->GetBaseId()).c_str(),dict->GetAt(lexUn[i]->GetBaseId()).length());
				counter+=dict->GetAt(lexUn[i]->GetBaseId()).length();

			}

			}
			std::cout<<"Node "<<myRank<<" prepared dictionary to be sent. "<<lexUn.size()<<" lexical units ready.\n";
			//Blocking till master is ready

			bool goFlag=false;
			while(!goFlag)
			{
				message=SENDING_DICT;
				MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);

				MPI_Recv(&message,1,MPI_INT,0,0,MPI_COMM_WORLD,stat);
				if(message==NOT_YET)
				{
					sleep(10);
				}
				else if(message==ALL_OK)
					goFlag=true;
				else
				{
					std::cout<<"Unknown error - message!= ALL_OK && message!=NOT_YET \n";
					exit(-1);
				}
			}
			std::cout<<"Node "<<myRank<<" received ALL_OK for dictionary from master\n";
			//Number of LUs
			message=numOfLexUnits;
			MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);

			//Size of each LU
			MPI_Send(tableLULength,numOfLexUnits,MPI_INT,0,0,MPI_COMM_WORLD);

			//TF for each LU
			MPI_Send(tableTF,numOfLexUnits,MPI_INT,0,0,MPI_COMM_WORLD);

			//Length of table
			message=counter;
			MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);

			//Table with chars
			MPI_Send(charTable,counter,MPI_CHAR,0,0,MPI_COMM_WORLD);
			delete [] charTable;
			delete [] tableOfIndexes;
			delete [] tableOfValues;


			me_architect::Config::Delete();
			//			Corpus::CWordDictionary::GetInstance()->Clear();
			//			LexicalUnits::UnigramDictionary::GetInstance()->Clear();
		}
		else //message!= ALL_OK && message!= EXIT_CODE
		{
			std::cout<<"Unknown error - message!= ALL_OK && message!= EXIT_CODE \n";
			exit(-1);
		}
	}

}
