/*
    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 "parallel_doc_term_architect.h"
#include "Config.h"
#include "MatrixSaver.h"
#include <boost/thread/thread.hpp>
#include <boost/ref.hpp>




namespace doc_term_architect
{
	DocTermArchitectMaster::DocTermArchitectMaster()
	{
		Config *cfg = Config::GetInstance();
		numOfJobs=cfg->m_vDocumentList.size();
		jobsFinished=0;
		jobIterator=0;
		results = new std::vector<Matrices::CompressedVectorPtr>();
		numOfColumns = cfg->m_vColumnTerms.size();
		tokenCounter=0;
	}

	void DocTermArchitectMaster::sendJob(int nodeNum)
	{
		if(jobIterator<numOfJobs)
		{
			mpiHandler.sendItem(jobIterator,MPI_INT,nodeNum);
			jobIterator++;
		}
		else
			mpiHandler.sendItem(EXIT_CODE,MPI_INT,nodeNum);
	}
	void DocTermArchitectMaster::receiveResults(int nodeNum)
	{
		int numOfJobsProcessedOnNode;
		mpiHandler.recvItem(numOfJobsProcessedOnNode,MPI_INT,nodeNum);
		jobsFinished+=numOfJobsProcessedOnNode;
		int tokensProcessedOnNode;
		mpiHandler.recvItem(tokensProcessedOnNode,MPI_INT,nodeNum);
		tokenCounter+=tokensProcessedOnNode;
		int numOfRows;
		mpiHandler.recvItem(numOfRows,MPI_INT,nodeNum);
		if(numOfRows>0)
		{
			int labelTablelength;
			char* labelTable = NULL;
			mpiHandler.recvArray(labelTable,labelTablelength,MPI_CHAR,nodeNum);
			int* labelLengthTable=NULL;
			mpiHandler.recvArray(labelLengthTable,numOfRows,MPI_INT,nodeNum);

			int *numOfValuesInRows=NULL;
			mpiHandler.recvArray(numOfValuesInRows,numOfRows,MPI_INT,nodeNum);
			int numOfNonZero;
			mpiHandler.recvItem(numOfNonZero,MPI_INT,nodeNum);
			int *tableOfIndexes=NULL;
			double *tableOfValues=NULL;

			mpiHandler.recvArray(tableOfIndexes,numOfNonZero,MPI_INT,nodeNum);
			mpiHandler.recvArray(tableOfValues,numOfNonZero,MPI_DOUBLE,nodeNum);

			int counter=0;
			for(int i=0;i<numOfRows;i++)
			{
				Matrices::CompressedVectorPtr row (new Matrices::CompressedVector(numOfColumns));
				for(int j=0;j<numOfValuesInRows[i];j++)
				{

					row->SetElement(tableOfIndexes[counter],tableOfValues[counter]);
					counter++;
				}
				results->push_back(row);

			}
			if(counter!=numOfNonZero)
				throw std::runtime_error("Error - counter!=numOfNonZero\n");

			counter=0;
			for(int i=0;i<numOfRows;i++)
			{
				char* buffer = new char[labelLengthTable[i]+1];
				strncpy(buffer,&(labelTable[counter]),labelLengthTable[i]);
				counter+=labelLengthTable[i];
				buffer[labelLengthTable[i]]='\0';
				rowLabels.push_back(std::string(buffer));

				delete buffer;
			}
			if(counter!=labelTablelength)
				throw std::runtime_error("Error: counter!=labelTablelength");


			delete labelTable;
			delete labelLengthTable;
			delete numOfValuesInRows;
			delete tableOfIndexes;
			delete tableOfValues;
		}
	}

	void DocTermArchitectMaster::Finalize()
	{
		Matrices::CRSMatrixPtr crs(new Matrices::CRSMatrix("crs",results,numOfColumns));
		MatrixSaver saver;
		saver.SaveToSuperMatrix(crs, tokenCounter, rowLabels);

	}

	void DocTermArchitectMaster::Start()
	{
		std::cout<<"Num of jobs: "<<numOfJobs<<std::endl;
		while(jobsFinished!=numOfJobs)
		{
			int message;
			int sourceNode=mpiHandler.recvItem(message,MPI_INT);
			if(message==GIVE_JOB)
				sendJob(sourceNode);
			else if(message==SENDING_RESULTS)
				receiveResults(sourceNode);
			else
				throw std::runtime_error("Received unknown message from node!\n");
		}
	}
	DocTermArchitectSlave::DocTermArchitectSlave()
	{
		tasksDone=0;
		m_stTokenCount=0;
		exitFlag=false;
	}

	int DocTermArchitectSlave::fetchJob()
	{
		int jobNumber;
		mpiHandler.sendItem(GIVE_JOB,MPI_INT,0);
		mpiHandler.recvItem(jobNumber,MPI_INT,0);
		if(jobNumber==EXIT_CODE)
		{
			exitFlag=true;
			jobNumber=-1;
		}
		return jobNumber;

	}
	void DocTermArchitectSlave::processToken(Corpus::CToken *token,Matrices::CompressedVectorPtr &document_vector )
	{
		Config* cfg = Config::GetInstance();
		int baseId = token->DisambLexem().GetBaseIdx();
		Corpus::TFlexClass flex = token->DisambLexem().GetTag().GetFlex();
		std::map<Corpus::TFlexClass, std::map<int, size_t> >::iterator find = cfg->m_mTermIndex.find(flex);
		if( find == cfg->m_mTermIndex.end() )
		{
			Corpus::Tokenisation::CTokenReader::Free( token );
			return;
		}
		std::map<int, size_t>::iterator inner_find = (find->second).find(baseId);
		if(inner_find == (find->second).end() )
		{
			Corpus::Tokenisation::CTokenReader::Free( token );
			return;
		}
		size_t column_no = inner_find->second;
		document_vector->SetElement( column_no, (*document_vector)[column_no] + 1);
		Corpus::Tokenisation::CTokenReader::Free( token );
	}

	void DocTermArchitectSlave::processDocument(int docNum)
	{
		Config* cfg = Config::GetInstance();
		std::string &doc = cfg->m_vDocumentList[ docNum ];
		std::ifstream file ( doc.c_str() );
		if ( !file )
		{
			std::cerr << "Cannot open " << doc << ". Skipping...";
			return;
		}
		Corpus::Tokenisation::CCorpusTokenReader reader ( file,true );
		//for each term
		Corpus::CToken *token = NULL;
		Matrices::CompressedVectorPtr document_vector (new Matrices::CompressedVector(cfg->m_vColumnTerms.size()));
		while( (token = reader.GetNext()) != NULL)
		{
			m_stTokenCount++;
			processToken(token, document_vector);
		}// for each term
		rowLabels.push_back(doc);
		results.push_back(document_vector);


	}
	std::vector<Corpus::CToken*> DocTermArchitectSlave::FlattenDocument(std::vector< boost::shared_ptr<Corpus::CSentence> > * pSentences)
	{
		Config* cfg=Config::GetInstance();
		std::vector<Corpus::CToken*> m_vDocument;
		for(size_t sent_no = 0; sent_no < pSentences->size(); sent_no++)
		{
			boost::shared_ptr<Corpus::CSentence> sent = (*pSentences)[sent_no];
			if((size_t)sent->Size()<=cfg->query_size)
				for(int tok_no = 0; tok_no < sent->Size(); tok_no++)
					m_vDocument.push_back( (*sent)[tok_no]->CreateHardCopy() );
		}
		return m_vDocument;
	}
	void DocTermArchitectSlave::processCorpus(int corpusNum)
	{
		Config* cfg = Config::GetInstance();
		PoliqarpTools::CPoliqarpReader cpr(cfg->m_vDocumentList[corpusNum], false, false);

		std::stringstream query;
		query << "[]{0," << cfg->query_size << "} within s";

		if(!cpr.CompileQuery(query.str().c_str() ) )
			throw std::runtime_error("Failed to compile querry");
		std::cout<<"Compiled query...\n";
		std::vector< boost::shared_ptr<CSentence> > * sentencesInDoc;
		while((sentencesInDoc = cpr.GetNextDocument()))
		{
			Matrices::CompressedVectorPtr document_vector (new Matrices::CompressedVector(cfg->m_vColumnTerms.size()));
			std::vector<Corpus::CToken*> tokensInSentence = FlattenDocument(sentencesInDoc);
			for(size_t i=0;i<tokensInSentence.size();i++)
				processToken(tokensInSentence[i],document_vector);
			//TODO: extracting document's name from poliqarp
			std::stringstream ss;
			ss<<cfg->m_vDocumentList[corpusNum]<<"_todo_"<<results.size();
			rowLabels.push_back(ss.str());
			results.push_back(document_vector);
		}
	}

	void DocTermArchitectSlave::Start()
	{
		while(!exitFlag)
		{
			int currentTask = fetchJob();
			if(currentTask!=-1)
			{
				std::cout<<"Received task no "<<currentTask<<"...\n";
				if(Config::GetInstance()->binaryDocuments)
					processCorpus(currentTask);
				else
					processDocument(currentTask);
				tasksDone++;
			}

		}
		sendResults();
	}

	void DocTermArchitectSlave::sendResults()
	{
		int labelTableLength=0;
		int* labelLengthTable=new int[rowLabels.size()];
		int* numOfValuesInRows = new int[results.size()];
		int numOfNonZero=0;
		char* labelTable = NULL;
		int *indexTable= NULL;
		double *valueTable = NULL;
		if(results.size()>0)
		{
			for(size_t i=0;i<rowLabels.size();i++)
			{
				labelTableLength+=rowLabels[i].length();
				labelLengthTable[i]=rowLabels[i].length();
				numOfValuesInRows[i]=results[i]->GetValues()->size();
				numOfNonZero+=numOfValuesInRows[i];
			}
			labelTable = new char[labelTableLength];
			indexTable= new int[numOfNonZero];
			valueTable = new double[numOfNonZero];

			int counter=0;
			int indexTableCounter=0;
			for(size_t i=0;i<rowLabels.size();i++)
			{
				memcpy(&(labelTable[counter]),rowLabels[i].c_str(),rowLabels[i].length());
				counter+=rowLabels[i].length();
				for(Matrices::STDoubleMapIter iter=results[i]->GetValues()->begin(); iter!=results[i]->GetValues()->end();++iter )
				{
					indexTable[indexTableCounter]=iter->first;
					valueTable[indexTableCounter]=iter->second;
					indexTableCounter++;
				}
			}
			if(counter!=labelTableLength)
				throw std::runtime_error("Something wrong: counter!=labelTableLength!\n");
			if(indexTableCounter!=numOfNonZero)
				throw std::runtime_error("Something wrong: indexTableCounter!=numOfNonZero!\n");
		}
		mpiHandler.sendItem(SENDING_RESULTS,MPI_INT,0);
		mpiHandler.sendItem(tasksDone,MPI_INT,0);
		mpiHandler.sendItem(m_stTokenCount,MPI_INT,0);
		mpiHandler.sendItem((int)results.size(),MPI_INT,0);
		if(results.size()>0)
		{
			mpiHandler.sendArray(labelTable,labelTableLength,MPI_CHAR,0);
			mpiHandler.sendArray(labelLengthTable,results.size(),MPI_INT,0);

			mpiHandler.sendArray(numOfValuesInRows,results.size(),MPI_INT,0);
			mpiHandler.sendItem(numOfNonZero,MPI_INT,0);
			mpiHandler.sendArray(indexTable,numOfNonZero,MPI_INT,0);
			mpiHandler.sendArray(valueTable,numOfNonZero,MPI_DOUBLE,0);
		}
	}

	DocTermMatrixBuilder::DocTermMatrixBuilder(std::vector<size_t> documents, Matrices::CRSMatrixPtr pMatrix)
		: m_pMatrix(pMatrix),
		m_vMyDocuments(documents)
	{
		m_iThreadNo = thread_counter ++;
		m_stTokenCount = 0;
	}

	int DocTermMatrixBuilder::thread_counter = 1;
	boost::mutex DocTermMatrixBuilder::cout_mutex;

	size_t DocTermMatrixBuilder::GetTokenCount()
	{
		return m_stTokenCount;
	}

	void DocTermMatrixBuilder::operator() ()
	{
		{
			boost::mutex::scoped_lock lock(cout_mutex);
			std::cout << "[" << m_iThreadNo  << "] starting with " << m_vMyDocuments.size() << " documents." <<  std::endl;
		}
		Config *c = Config::GetInstance();
		//for each document
		for(size_t doc_no = 0; doc_no < m_vMyDocuments.size(); doc_no++)
		{
			std::string &doc = c->m_vDocumentList[ m_vMyDocuments[doc_no] ];
			std::ifstream file ( doc.c_str() );
			if ( !file )
			{
				std::cerr << "Cannot open " << doc << ". Skipping...";
				continue;
			}
			Corpus::Tokenisation::CCorpusTokenReader reader ( file,true );
			//for each term
			Corpus::CToken *token = NULL;
			Matrices::CompressedVectorPtr document_vector = m_pMatrix->GetCompressedRow( m_vMyDocuments[doc_no] );
			while( (token = reader.GetNext()) != NULL)
			{
				m_stTokenCount++;
				int baseId = token->DisambLexem().GetBaseIdx();
				Corpus::TFlexClass flex = token->DisambLexem().GetTag().GetFlex();
				std::map<Corpus::TFlexClass, std::map<int, size_t> >::iterator find = c->m_mTermIndex.find(flex);
				if( find == c->m_mTermIndex.end() )
				{
					Corpus::Tokenisation::CTokenReader::Free( token );
					continue;
				}
				std::map<int, size_t>::iterator inner_find = (find->second).find(baseId);
				if(inner_find == (find->second).end() )
				{
					Corpus::Tokenisation::CTokenReader::Free( token );
					continue;
				}
				size_t column_no = inner_find->second;
				document_vector->SetElement( column_no, (*document_vector)[column_no] + 1);
				Corpus::Tokenisation::CTokenReader::Free( token );

			}// for each term

			if( doc_no % 100 == 0 )
			{
				boost::mutex::scoped_lock lock(cout_mutex);
				std::cout << "[" << m_iThreadNo << "] Document " << doc_no << " / " << m_vMyDocuments.size()  << std::endl;
			}
		}//for each doc

		{
			boost::mutex::scoped_lock lock(cout_mutex);
			std::cout << "[" << m_iThreadNo  << "] finished" << std::endl;
		}
	}// operator()


}//namespace doc_term_architect
