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

namespace parallel
{
	namespace similarity
	{
		int partition(double* array,int*indexArray, int left, int right, int pivotIndex)
		{
			double pivotValue = array[pivotIndex];
			std::swap(array[pivotIndex],array[right]);
			std::swap(indexArray[pivotIndex],indexArray[right]);
			int storeIndex=left;
			for(int i=left;i<right;i++)
			{
				if(array[i] >=pivotValue)
				{
					std::swap(array[i],array[storeIndex]);
					std::swap(indexArray[i],indexArray[storeIndex]);
					storeIndex++;
				}
			}
			std::swap(array[storeIndex],array[right]);
			std::swap(indexArray[storeIndex],indexArray[right]);
			return storeIndex;
		}

		void quicksort(double* array,int* indexArray, int left, int right)
		{
			if(right>left)
			{
				int pivotIndex=(left+right)/2;
				pivotIndex=partition(array,indexArray,left,right,pivotIndex);
				quicksort(array,indexArray,left,pivotIndex-1);
				quicksort(array,indexArray,pivotIndex+1,right);
			}
		}

		MatrixSaverInterface* loadMatrix(int numOfRows)
		{
			MatrixSaverInterface* result=NULL;
			Config* cfg = Config::getInstance();
			std::string matrixType = cfg->typeOfMatrix;

			if (matrixType.compare("supermatrix")==0)
			{
				result = new MatrixSaverSuperMatrix(numOfRows);
				//
			}
			else if(matrixType.compare("textfile")==0)
				result = new MatrixSaverTextFile();
			else if(matrixType.compare("densematrix")==0)
				result = new MatrixSaverDenseMatrix(numOfRows);

			else
			{

				throw std::runtime_error("Wrong results type");
			}

			return result;

		}
		void ParallelMaster::printETA()
		{
			using namespace boost::posix_time;
			//ETA
			ptime end = microsec_clock::local_time();
			time_duration len = end - start;
			if( len.total_milliseconds()> 5000)
			{
				total_time += len;
				long total_time_ms = total_time.total_milliseconds();
				long eta_ms=-1;
				if(numOfComputedRows!=0)
					eta_ms = (total_time_ms * (numOfRowsToCompute-numOfComputedRows) ) / numOfComputedRows;
				std::cout << "Processed " << numOfComputedRows<< " / " << numOfRowsToCompute
						<< " Elapsed: " << milliseconds(total_time_ms) 	<< " ETA: " << to_simple_string( milliseconds(eta_ms) )
						<< "            \r";
				std::cout.flush();
				start = microsec_clock::local_time();
			}//if len > 5000
		}

		ParallelMaster::ParallelMaster()
		{
		}
		ParallelMaster::~ParallelMaster()
		{
			if(isResult!=NULL)
				delete isResult;
		}

		void ParallelMaster::LoadWordsList(Matrices::SuperMatrixPtr matrix)
		{
			Config* cfg=Config::getInstance();


			if(cfg->wordsFile=="")
			{
				for(int i=0;i<totalNumOfRows;i++)
					isResult[i]=false;
				numOfRowsToCompute=totalNumOfRows;
			}
			else
			{
				numOfRowsToCompute=0;
				std::string word;
				for(int i=0;i<totalNumOfRows;i++)
					isResult[i]=true;
				std::ifstream file(cfg->wordsFile.c_str() );
				while(getline(file,word))
				{
					size_t index = matrix->GetRowIndex(word);
					if(index<matrix->NumberOfRows())
					{
						isResult[matrix->GetRowIndex(word)]=false;
						numOfRowsToCompute++;
					}
				}
				file.close();
			}
		}

		void ParallelMaster::Init(MPI_Status stat)
		{
			this->stat=stat;
			Config* cfg=Config::getInstance();

			//Creation of compare function to obtain similarity of measure
			Matrices::SuperMatrixPtr matrix = Matrices::IO::LoadSuperMatrix(cfg->dir,cfg->matrixName,cfg->ccs,cfg->crs);
			smartcomparator::compare::CompareFunctionPtr pCmp = smartcomparator::compare::FunctionFactory::CreateFunction(cfg->function,matrix);

			rowIterator=0;
			numOfComputedRows=0;
			totalNumOfRows=matrix->NumberOfRows();
			MPI_Comm_size(MPI_COMM_WORLD,&numOfNodes);

			resultsMatrix = loadMatrix(totalNumOfRows);
			resultsMatrix->initialize();

			isResult = new bool[totalNumOfRows];
			LoadWordsList(matrix);
			isSimetrical = resultsMatrix->isSimetrical(pCmp->IsSymmetrical());
			matrix.reset();
			pCmp.reset();

			start = boost::posix_time::microsec_clock::local_time();
		}

		void ParallelMaster::receivedRowRequest(int nodeNum)
		{
			int message;
			while(rowIterator<totalNumOfRows && isResult[rowIterator])
			{
				rowIterator++;
			}
			if(rowIterator==totalNumOfRows)
			{
				message=EXIT_CODE;
				mpiHandler.sendItem(message,MPI_INT,nodeNum);
			}
			else
			{
				message=rowIterator;
				mpiHandler.sendItem<int>(message,MPI_INT,nodeNum);
				rowIterator++;
			}
		}
		void ParallelMaster::receivingResults(int nodeNum)
		{
			int message=ALL_OK;
			mpiHandler.sendItem(message,MPI_INT,nodeNum);

			int numOfReceivedRow;
			mpiHandler.recvItem(numOfReceivedRow,MPI_INT,nodeNum);
			int *resultsIndexes=NULL;
			double* results=NULL;
			int numOfResults;
			mpiHandler.recvArray(resultsIndexes,numOfResults,MPI_INT,nodeNum);
			mpiHandler.recvArray(results,numOfResults,MPI_DOUBLE,nodeNum);
			mpiHandler.sendItem(message,MPI_INT,nodeNum);

			isResult[numOfReceivedRow]=true;
			numOfComputedRows++;
			resultsMatrix->addValues(numOfReceivedRow,numOfResults,resultsIndexes,results);

			delete resultsIndexes;
			delete results;

		}

		void ParallelMaster::Start()
		{
			int message;
			MPI_Request request;
			int sourceNode;
			int probeFlag=0;

			while(numOfComputedRows<numOfRowsToCompute)
			{
				probeFlag = mpiHandler.probeForMessage();
				while(probeFlag==0)
				{
					printETA();
					usleep(100);
					probeFlag = mpiHandler.probeForMessage();
				}
				printETA();
				sourceNode=mpiHandler.recvItem(message,MPI_INT);
				if(message==GIVE_ME_ROW)
				{
					receivedRowRequest(sourceNode);
				}
				else if(message==SENDING_RESULTS)
				{
					receivingResults(sourceNode);
				}
			}
			message=EXIT_CODE;
			for(int i=1;i<numOfNodes;i++)
			{
				MPI_Isend(&message,1,MPI_INT,i,0,MPI_COMM_WORLD,&request);
				MPI_Request_free(&request);
			}
			resultsMatrix->finalize();
		}
		ParallelSlave::ParallelSlave(int myRank)
		{
			this->myRank=myRank;
		}
		ParallelSlave::~ParallelSlave()
		{
			if(results)
				delete results;
			if(resultsIndex)
				delete resultsIndex;
		}

		void ParallelSlave::Init(MPI_Status stat)
		{
			this->stat = stat;
			Config* cfg= Config::getInstance();
			//Loading matrix
			Matrices::SuperMatrixPtr matrix = Matrices::IO::LoadSuperMatrix(cfg->dir,cfg->matrixName,cfg->ccs,cfg->crs);

			//Creation of compare function
			pCmp = smartcomparator::compare::FunctionFactory::CreateFunction(cfg->function,matrix);
			if(cfg->WeightMatrix)
			{
				pCmp->WeightMatrix();


			}

			numOfAllRows = matrix->NumberOfRows();

			MatrixSaverInterface* resultsMatrixInterface = loadMatrix(numOfAllRows);

			isSimetrical=resultsMatrixInterface->isSimetrical(pCmp->IsSymmetrical());
			delete resultsMatrixInterface;

			resultsIndex = new int[numOfAllRows];
			results = new double[numOfAllRows];


		}

		int ParallelSlave::Compute()
		{
			Config* cfg= Config::getInstance();
			int rowIterator;
			if(isSimetrical)
			{
				rowIterator=currentRowNum+1;
			}
			else rowIterator = 0;

			int resultsIterator;
			for(resultsIterator=0;rowIterator<numOfAllRows;rowIterator++)
			{
				double result=pCmp->Compare(currentRowNum,rowIterator);
				if(result> cfg->minimalSimilarity)
				{
					resultsIndex[resultsIterator]=rowIterator;
					results[resultsIterator++]=result;
				}
			}
			return resultsIterator;
		}
		void ParallelSlave::sendResults(int numOfResults)
		{
			MPI_Request request;
			int message;
			message = SENDING_RESULTS;
			MPI_Isend(&message,1,MPI_INT,0,0,MPI_COMM_WORLD,&request);
			MPI_Request_free(&request);

			//Receiving ALL_OK
			mpiHandler.recvItem(message,MPI_INT,0);

			//Sending row number
			mpiHandler.sendItem(currentRowNum,MPI_INT,0);

			//Sending table with indexes
			mpiHandler.sendArray(resultsIndex,numOfResults,MPI_INT,0);

			//Sending table with results
			mpiHandler.sendArray(results,numOfResults,MPI_DOUBLE,0);

			//Receiving of ALL_OK
			mpiHandler.recvItem(message,MPI_INT,0);
			if(message!= ALL_OK)
			{
				std::cout<<"Unknown error - message!= ALL_OK \n";
				exit(-1);
			}
		}

		void ParallelSlave::Start()
		{
			Config* cfg= Config::getInstance();
			MPI_Request request;
			int message;
			bool exitFlag = false;
			while(!exitFlag)
			{
				message = GIVE_ME_ROW;
				MPI_Isend(&message,1,MPI_INT,0,0,MPI_COMM_WORLD,&request);
				MPI_Request_free(&request);

				//Receiving row to compute
				mpiHandler.recvItem(currentRowNum,MPI_INT,0);
				if(currentRowNum==EXIT_CODE)
				{
					exitFlag=true;
					break;
				}

				//Computing
				int numOfResults = Compute();

				//Sorting results
				if(numOfResults>cfg->kBestResults)
				{
					quicksort(results,resultsIndex,0,numOfResults-1);
					numOfResults=cfg->kBestResults;
				}

				//Sending results back
				sendResults(numOfResults);


			}
		}

		void runParallelSimilarityFunction(int ac, char **av)
		{

			int myRank;
			MPI_Status stat;
			MPI_Init(&ac,&av);

			MPI_Comm_rank(MPI_COMM_WORLD,&myRank);
			Config* cfg = Config::getInstance();

			if(myRank==0)
			{
				std::string reportFilename=cfg->resultsFile+"_report";
				std::fstream reportFile(reportFilename.c_str(),std::ios::out);
				int numOfNodes;
				MPI_Comm_size(MPI_COMM_WORLD,&numOfNodes);
				std::cout << "Nodes: " << numOfNodes << std::endl;

				double startTime=MPI_Wtime();
				ParallelMaster master;
				master.Init(stat);
				master.Start();
				std::cout<<"Master finished.\n";
				double totalTime=MPI_Wtime()-startTime;
				reportFile<<"Matrix: "<<cfg->matrixName<<std::endl;
				reportFile<<"Similarity function: "<<cfg->function<<std::endl;
				reportFile<<"Number of nodes: "<<numOfNodes<<std::endl;
				reportFile<<"Minimal similarity: "<<cfg->minimalSimilarity<<std::endl;
				reportFile<<"K-best results: "<<cfg->kBestResults<<std::endl;
				reportFile<<"Time taken: "<<totalTime<<std::endl;
				reportFile.close();

			}
			else
			{
				ParallelSlave slave(myRank);
				slave.Init(stat);
				slave.Start();
				std::cout<<"Slave "<<myRank<<" finished.\n";
			}

			MPI_Finalize();
		}

	}
}
