/***************************************************************************
                Job Manager
                -----------------------------------------
	begin      : 2012/05/14
	copyright  : (C) 2011 SolutionBox Inc.
	author     : Service 1 Team
	email      : svc1@solbox.com
	version    : 3.1.0

    CopyRight(C) 2011 SolutionBox Inc. All Rights reserved.
    Redistribution and use in source and binary forms, with or with out
    modification, are not permitted in outside of SolutionBox Inc.
***************************************************************************/
#include "rc_cmove.h"
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fstream>
#include "jobmanager.h"
#include "parameter.h"
#include "workpool.h"
#include "Logger.h"
#include "dbconnpool.h"

#define  WORK_WAIT		1000000 // microsecond

static int worknotifyfn(void *object , short success)
{	jobmanager* manager = reinterpret_cast<jobmanager *>(object);
	ostringstream msg;
	msg << "Called worknotifyfn :" << success;
	_LOGACOUT(LDEV2, msg);
	return manager->worknotify(success);
}

jobmanager::jobmanager(JOB_TYPE::JOB_TYPE type, const char* syncpath)
	: m_type(type),m_exit(false)
{
	m_tmppath.m_syncpath = syncpath;
	pthread_mutex_init(&m_notifymutex, NULL);
}

jobmanager::~jobmanager()
{
	pthread_mutex_destroy(&m_notifymutex);
}

int jobmanager::run()
{

	int nRet = pthread_create(&m_hthread, 0, jobmanager::WorkFn, this);
	if( nRet )
	{
		ostringstream msg;
		msg << "Thread create failed.[" << errno << "]";
		LOGACERR(LERR, msg);

		return -1;
	}
	return 0;
}

bool jobmanager::gettargethostlist(string path, vector<string> &vechost)
{
	ostringstream msg;
	if(path.size() > 0 )
	{
		ifstream hostfile;
		hostfile.open(path.c_str());
		if( hostfile.is_open() == false )
		{
			msg << "Target Host file open error.[" << path << "]";
			LOGACERR(LERR, msg);
			return false;
		}

		string line;
		while (!hostfile.eof() )
		{
			getline (hostfile,line);
			if(line.empty() == true || line == "\r")
			{
#ifdef _DEBUG
				cout << "host data read : empty or linefeed" << line << endl;
#endif // _DEBUG
				continue;
			}

			// check  IP addres
			struct in_addr laddr;
			int status = inet_aton(line.c_str(), &laddr);
			if(status != 0)
			{
				msg << "Target Host Not DNS Name.['" << line  << "' IN " << path << "]";
				LOGACERR(LERR, msg);
				continue;
			}

			vechost.push_back(line);
		}

		hostfile.close();
		if( vechost.size() == 0 )
		{
			msg << "Target Host file Empty.[" << path << "]";
			LOGACERR(LERR, msg);
			return false;
		}
	}
	return true;
}

bool jobmanager::runsync()
{
	ostringstream msg;
	ifstream syncfile;
	
	msg << "Job - Synchronize working...";
	_LOGACOUT(LINF, msg);

	// target host list
	vector<string> hostlist;
	if( gettargethostlist(parameter::getInstance()->gettargetlistpath(), hostlist) == false)
	{
		return false;
	}
	
	// get sync file line
	m_status.m_diff = getfileline(m_tmppath.m_syncpath.c_str());
	
	syncfile.open(m_tmppath.m_syncpath.c_str());

	if( syncfile.is_open() == false )
	{
		msg << "Sync data file open error.[" << m_tmppath.m_syncpath <<"]";
		LOGACERR(LERR, msg);
		return false;
	}

	string line;
	while (!syncfile.eof() )
	{
		getline (syncfile,line);
		if(line.empty() == true || line == "\r")
		{
#ifdef _DEBUG
			cout << "Sync data read : empty or linefeed" << line << endl;
#endif // _DEBUG
			if(!syncfile.eof())
			{
				++m_status.m_launcherfail;
				msg << "File Data Empty. or carriage return." << line;
				LOGACERR(LERR, msg);
			}
			continue;
		}

		if(m_exit)
		{
			msg << "Work force to end.";
			LOGACERR(LNOT, msg);
			break;
		}

#ifdef _DEBUG
		cout << "Sync data read : " << line << endl;
#endif // _DEBUG
		work *w = workpool::getInstance()->GetWorkPool();
		if(w != NULL )
		{
			++m_status.m_work;
			w->setenablesync("1111");
			w->setnetworkmode(parameter::getInstance()->getnetworkmode());
			if( hostlist.size() > 0 )
				w->settargethost(hostlist);
			w->run(m_type,line,worknotifyfn, this);
		}
		else
		{
			++m_status.m_launcherfail;
			msg << "Get Work Pool error :" << line;
			LOGACERR(LERR, msg);
		}
	}

	// 마지막 라인에 캐리지 리턴 없는 경우 파일을 실제 읽었을 때보다 
	// 한라인 작은 수를 wc -l 에서 출력되어 작업이 완료되지 않는 것을
	// 막기 위해서 work가 diff 값보다 클 때 diff 값을 work 값으로 보정
	if (m_status.m_diff < m_status.m_work)
		m_status.m_diff = m_status.m_work;
		
	syncfile.close();

	msg << "Job - Synchronize working End";
	_LOGACOUT(LINF, msg);
	return true;
}

void jobmanager::cleandata()
{
	/*
	// delete temp file
#ifndef _DEBUG
	unlink(m_tmppath.m_msrcpath.c_str());
	unlink(m_tmppath.m_ssrcpath.c_str());
	unlink(m_tmppath.m_diffpath.c_str());
	unlink(m_tmppath.m_syncpath.c_str());
#endif // _DEBUG
	
	// delete memory
	while( m_srcdata.size() > 0 )
	{
		vector<CSourceData*>::iterator iter = m_srcdata.begin();
		m_srcdata.erase(iter);
		delete static_cast<CSourceData *>(*iter);
	}
	*/
}

void jobmanager::cleanup( void* pdata )
{
	jobmanager* pObject = reinterpret_cast<jobmanager *>(pdata);
#ifdef _DEBUG
	cout << "Job cleanup thread:" << pthread_self() << endl;
#endif // _DEBUG
	pObject->cleandata();
}

void* jobmanager::WorkFn( void* pdata )
{
	jobmanager* pObject = reinterpret_cast<jobmanager *>(pdata);

	ostringstream msg;
	 msg << "Job Manager thread start [" << "???" << "]";
	 _LOGACOUT(LDEV2, msg);

	pthread_cleanup_push(jobmanager::cleanup, pdata);

	bool running = true;
	while(running)
	{
		// sync
		pObject->runsync();

		// waiting
		pObject->waiting();

		running = false;
	}

	msg << "Job Manager thread end[" << "???" << "]";
	_LOGACOUT(LDEV2, msg);

	pthread_cleanup_pop(1);

	return 0;
}

int jobmanager::worknotify(short success)
{
	ostringstream msg;
	pthread_mutex_lock(&m_notifymutex);
		
	msg << "work notify :" << success;
	_LOGACOUT(LDEV2, msg);

	if (success == 0)
		++m_status.m_success;
	else 
		++m_status.m_error;

	if(success != 0)
	{
		if( masterdbpool::getInstance()->GetPoolSize() == 0 ) 
		{
			msg << "Master DB Pool empty. And Work force to end.";
			LOGACERR(LDEV1, msg);
			m_exit = true;
		}
	}

	pthread_mutex_unlock(&m_notifymutex);

	return 0;
}

int64_t jobmanager::getfileline(const char* path)
{
	int64_t r = 0;
	ostringstream cmd, msg;

	cmd << "cat " << path << " | wc -l";
	string lines = getCommand(cmd.str());

	//msg << "file lines : CMD - " << cmd.str() << " => " << lines;
	//_LOGACOUT(LDBG,msg);
	
	r = atoll(lines.c_str());
	msg << "file lines : " << path << " => " << r;
	_LOGACOUT(LDBG,msg);

	return r;
}

bool jobmanager::waiting()
{
	bool r = true;
	ostringstream msg;
	do
	{
		if (m_status.m_diff ==
			(m_status.m_error + m_status.m_success + m_status.m_launcherfail) )
		{			
			msg << "work end of waiting";
			_LOGACOUT(LDEV2, msg);
			printstatus();
			break;
		}

		if( (m_exit) &&(m_status.m_work == (m_status.m_error + m_status.m_success)) ) 
		{			
			msg << "Work force to end.";
			LOGACERR(LNOT, msg);
			break;
		}

#ifdef _DEBUG 
		msg << "work of waiting.........";
		_LOGACOUT(LDBG, msg);
		printstatus();
#endif // _DEBUG 

		solusleep(WORK_WAIT);
	}while(true);
	
	return r;
}

void jobmanager::printstatus(string prefixed)
{
	ostringstream msg;
	if( prefixed.empty() == false )
		msg << "[" << prefixed << "]";

	msg << "Job Manager(" << "???" << "=>" << "!!!" << 
		") total : " << m_status.m_diff << "(R" << m_status.m_work << "/F" <<
		m_status.m_launcherfail <<"), result :S" << m_status.m_success << "/E" << m_status.m_error;

	_LOGACOUT(LINF, msg);
}
