/*
    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 "parallelhal.h"

static log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("SuperMatrix.parallel.hal_architect.parallelhal"));

namespace parallel
{
	namespace hal_architect
	{
		ParallelHalMaster::ParallelHalMaster()
		{
		}
		void ParallelHalMaster::Init()
		{
			::hal_architect::Config *cfg= ::hal_architect::Config::GetInstance();
			MPI_Comm_size(MPI_COMM_WORLD,&numOfNodes);
			tasksGiven=0;
			totalTasks=cfg->m_sCorpusFiles.size();
			resultsReceived=0;
			corpusSize=-1;
			arch.Initalize();
		}
		void ParallelHalMaster::Finalize()
		{
			::hal_architect::MatrixSaver ms;
			::hal_architect::Config *config = ::hal_architect::Config::GetInstance();
			ms.SaveToSuperMatrix(  arch.getMatrix(), corpusSize, wordTF);
			//LexicalUnits::UnigramDictionary::GetInstance()->Save(config->workdir + config->m_sMatrixName +"/udict.txt");
			saveDict();

		}
		void ParallelHalMaster::saveDict()
		{
			::hal_architect::Config *config = ::hal_architect::Config::GetInstance();
			std::ofstream udictFile((config->workdir + config->m_sMatrixName +"/udict.txt").c_str());
			if(!udictFile)
				LOG4CXX_INFO(logger,"Cannot open udict.txt");

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

		void ParallelHalMaster::sendJob(int nodeNum)
		{
			if(tasksGiven<totalTasks)
			{
				mpiHandler.sendItem(tasksGiven,MPI_INT,nodeNum);
				tasksGiven++;
			}
			else
			{
				int message=EXIT_CODE;
				mpiHandler.sendItem(message,MPI_INT,nodeNum);
			}
		}
		void ParallelHalMaster::receiveResults(int nodeNum)
		{
			int processedCorpuses;
			mpiHandler.recvItem(processedCorpuses,MPI_INT,nodeNum);
			Matrices::CRSMatrixPtr crsMatrix = arch.getMatrix();
			if(processedCorpuses>0)
			{
				int numOfRows;// = crsMatrix->NumberOfRows();
				int *numOfValuesInRows=NULL;// = new int[numOfRows];
				mpiHandler.recvArray(numOfValuesInRows,numOfRows,MPI_INT,nodeNum);
				//				MPI_Recv(numOfValuesInRows,numOfRows,MPI_INT,nodeNum,0,MPI_COMM_WORLD,stat);
				int numOfNonZero;
				mpiHandler.recvItem(numOfNonZero,MPI_INT,nodeNum);
				//				MPI_Recv(&numOfNonZero,1,MPI_INT,nodeNum,0,MPI_COMM_WORLD,stat);
				int *tableOfIndexes=NULL;
				double *tableOfValues=NULL;
				//				tableOfIndexes = new int[numOfNonZero];
				//				tableOfValues = new double[numOfNonZero];

				mpiHandler.recvArray(tableOfIndexes,numOfNonZero,MPI_INT,nodeNum);
				//				MPI_Recv(tableOfIndexes,numOfNonZero,MPI_INT,nodeNum,0,MPI_COMM_WORLD,stat);
				mpiHandler.recvArray(tableOfValues,numOfNonZero,MPI_DOUBLE,nodeNum);
				//				MPI_Recv(tableOfValues,numOfNonZero,MPI_DOUBLE,nodeNum,0,MPI_COMM_WORLD,stat);

				//Filling matrices
				int counter=0;
				for(int i=0;i<numOfRows;i++)
				{
					for(int j=0;j<numOfValuesInRows[i];j++)
					{

						crsMatrix->AddToCell(tableOfValues[counter],i,tableOfIndexes[counter]);
						counter++;
					}

				}
				if(counter!=numOfNonZero)
					throw std::runtime_error("Error - counter!=numOfNonZero\n");
				delete numOfValuesInRows;
				delete tableOfIndexes;
				delete tableOfValues;
				LOG4CXX_INFO(logger,"Received part of matrix from node "<<nodeNum);

				int message;
				mpiHandler.recvItem(message,MPI_INT,nodeNum);
				if(message!=SENDING_CORPUS_SIZE)
					throw std::runtime_error("Error! Unrecognized message, message!=SENDING_CORPUS_SIZE!\n");
				mpiHandler.recvItem(corpusSize,MPI_INT,nodeNum);
			}
		}
		void ParallelHalSlave::sendResults(::hal_architect::Architect arch)
		{
			Matrices::CRSMatrixPtr matrix = arch.getMatrix();
			int numOfRows=-1;
			int numOfNonZero=-1;
			int *numOfValuesInRows=NULL;
			int *tableOfIndexes=NULL;
			double *tableOfValues=NULL;
			if(corpusesProcessed>0)
			{

				numOfRows = matrix->NumberOfRows();
				numOfNonZero = 0;


				numOfValuesInRows = new int[numOfRows];

				Matrices::STDoubleMapPtr doubleMapPtr;
				for(int i=0;i<numOfRows;i++)
				{
					doubleMapPtr = matrix->GetCompressedRow(i)->GetValues();

					numOfValuesInRows[i]=doubleMapPtr->size();
					numOfNonZero+=doubleMapPtr->size();
				}

				tableOfIndexes= new int[numOfNonZero];

				tableOfValues = new double[numOfNonZero];
				int counter=0;
				for(int i=0;i<numOfRows;i++)
				{
					doubleMapPtr = matrix->GetCompressedRow(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)
				{
					LOG4CXX_ERROR(logger,"Uknown error - counter!=numOfNonZero");
					exit(-1);
				}
			}
			int message=SENDING_RESULTS;
			mpiHandler.sendItem(message,MPI_INT,0);
			mpiHandler.sendItem(corpusesProcessed,MPI_INT,0);
			if(corpusesProcessed>0)
			{

				mpiHandler.sendArray(numOfValuesInRows,numOfRows,MPI_INT,0);

				//				MPI_Send(numOfValuesInRows,numOfRows,MPI_INT,0,0,MPI_COMM_WORLD);

				mpiHandler.sendItem(numOfNonZero,MPI_INT,0);
				//				MPI_Send(&numOfNonZero,1,MPI_INT,0,0,MPI_COMM_WORLD);

				mpiHandler.sendArray(tableOfIndexes,numOfNonZero,MPI_INT,0);
				//				MPI_Send(tableOfIndexes,numOfNonZero,MPI_INT,0,0,MPI_COMM_WORLD);

				mpiHandler.sendArray(tableOfValues,numOfNonZero,MPI_DOUBLE,0);
				//				MPI_Send(tableOfValues,numOfNonZero,MPI_DOUBLE,0,0,MPI_COMM_WORLD);

				LOG4CXX_INFO(logger,"Node "<<myRank<<" sent part of matrix to master");

				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);

			}
		}


		void ParallelHalMaster::Start()
		{
			int message;

			while(resultsReceived<totalTasks)
			{
				int sourceNode=mpiHandler.recvItem(message,MPI_INT,-1);
				if(message==GIVE_JOB)
					sendJob(sourceNode);
				else if(message==SENDING_RESULTS)
					receiveResults(sourceNode);
				else if(message==SENDING_DICT)
					receiveDict(sourceNode);
				else
				{
					LOG4CXX_ERROR(logger,"Received message: "<<message);
					throw std::runtime_error("Unknown error - unrecognized message!\n");
				}
			}
			int counter=0;

			for(std::map<std::string, int>::iterator iter=wordTF.begin();iter!=wordTF.end();++iter)
			{
				if(iter->second>0)
					counter++;
			}
			LOG4CXX_INFO(logger,"Counted "<<counter<<" positive values.");
		}

		ParallelHalSlave::ParallelHalSlave()
		{

		}
		void ParallelHalSlave::Init(int myRank)
		{
			this->myRank=myRank;
		}
		void ParallelHalMaster::receiveDict(int nodeNum)
		{
			LOG4CXX_INFO(logger,"Receiving dict from node "<<nodeNum);
			int message;
			mpiHandler.recvItem(message,MPI_INT,nodeNum);
			//			MPI_Recv(&message,1,MPI_INT,nodeNum,0,MPI_COMM_WORLD,stat);
			int corpussesProcessed = message;
			if(corpussesProcessed>0)
			{
				mpiHandler.recvItem(message,MPI_INT,nodeNum);
				//				MPI_Recv(&message,1,MPI_INT,nodeNum,0,MPI_COMM_WORLD,stat);
				int numOfLUs=message;
				LOG4CXX_INFO(logger,"Received numOfLus: "<<numOfLUs);
				int* tableLULength = new int[numOfLUs];
				int* tableTF = new int[numOfLUs];
				int lengthCharTable;

				mpiHandler.recvArray(tableLULength,numOfLUs,MPI_INT,nodeNum);
				//				MPI_Recv(tableLULength,numOfLUs,MPI_INT,nodeNum,0,MPI_COMM_WORLD,stat);

				mpiHandler.recvArray(tableTF,numOfLUs,MPI_INT,nodeNum);
				//				MPI_Recv(tableTF,numOfLUs,MPI_INT,nodeNum,0,MPI_COMM_WORLD,stat);

				//				mpiHandler.recvItem(lengthCharTable,MPI_INT,nodeNum);
				//				MPI_Recv(&lengthCharTable,1,MPI_INT,nodeNum,0,MPI_COMM_WORLD,stat);

				char* charTable = NULL;//new char[lengthCharTable];
				mpiHandler.recvArray(charTable,lengthCharTable,MPI_CHAR,nodeNum);
				//				MPI_Recv(charTable,lengthCharTable,MPI_CHAR,nodeNum,0,MPI_COMM_WORLD,stat);

				int counter=0;
				for(int i=0;i<numOfLUs;i++)
				{
					char* 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;
				LOG4CXX_INFO(logger,"Received dictionary from node "<<nodeNum);
				resultsReceived+=corpussesProcessed;
			}
		}

		void ParallelHalSlave::sendDict()
		{
			int *tableTF=NULL;
			int *tableLULength=NULL;
			int counter=-1;
			char *charTable=NULL;
			int numOfLexUn=0;
			if(corpusesProcessed>0)
			{
				//Sending dictionary
				LexicalUnits::UnigramDictionary *udict = LexicalUnits::UnigramDictionary::GetInstance();
				Corpus::CWordDictionary *dict = Corpus::CWordDictionary::GetInstance();
				::hal_architect::Config *config = ::hal_architect::Config::GetInstance();

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


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

				int lengthSingleLU;

				int positiveCounter=0;
				for(size_t i=0;i<lexUn.size();i++)
				{
					if(udict->GetCount(lexUn[i]))
					{
						tableTF[numOfLexUn] = udict->GetCount(lexUn[i]);
						if(tableTF[numOfLexUn]>0)
							positiveCounter++;
						lengthSingleLU=0;
						lengthSingleLU+=strlen(Corpus::CTag::GetFlexClassString(lexUn[i]->GetFlexClass()));
						lengthSingleLU+=1; // ":"
						lengthSingleLU+=dict->GetAt(lexUn[i]->GetBaseId()).length();
						tableLULength[numOfLexUn]=lengthSingleLU;


						counter+=lengthSingleLU;
						numOfLexUn++;
					}
				}

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


				for(size_t i=0;i<lexUn.size();i++)
				{
					if(udict->GetCount(lexUn[i]))
					{
						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();
					}
				}
				LOG4CXX_INFO(logger,"Node "<<myRank<<" prepared dictionary to be sent. "<<lexUn.size()<<" lexical units ready.");
			}
			int message;
			message=SENDING_DICT;
			MPI_Send(&message,1,MPI_INT,0,0,MPI_COMM_WORLD);
			mpiHandler.sendItem(corpusesProcessed,MPI_INT,0);
			if(corpusesProcessed>0)
			{
				//Number of LUs
				message=numOfLexUn;
				mpiHandler.sendItem(message,MPI_INT,0);
				//Size of each LU
				mpiHandler.sendArray(tableLULength,numOfLexUn,MPI_INT,0);
				//TF for each LU
				mpiHandler.sendArray(tableTF,numOfLexUn,MPI_INT,0);
				//Length of table
				message=counter;
				//Table with chars
				mpiHandler.sendArray(charTable,counter,MPI_CHAR,0);
				LOG4CXX_INFO(logger,"Sent charTable");
			}
		}


		void ParallelHalSlave::Start()
		{
			::hal_architect::Architect arch;
			arch.Initalize();
			bool exitFlag=false;
			int message;
			corpusesProcessed=0;
			while(!exitFlag)
			{
				message=GIVE_JOB;
				mpiHandler.sendItem(message,MPI_INT,0);
				mpiHandler.recvItem(message,MPI_INT,0);
				if(message==EXIT_CODE)
					exitFlag=true;
				else
				{
					arch.Start(message);
					corpusesProcessed++;
				}
			}
			sendResults(arch);
			sendDict();

			::hal_architect::Config::Delete();

		}
	}
}
