/***************************************************************************
                Work Pool
                -----------------------------------------
	begin      : 2012/05/13
	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 "workpool.h"
#include "validation.h"
#include "synchronization.h"
#include "parameter.h"
#include "Logger.h"
#include "ReportLog.h"

#include <fcntl.h>           /* For O_* constants */
#include <sys/stat.h>        /* For mode constants */
#ifdef _USE_POSIX_SEMAPHORES
#include <semaphore.h>
#elif defined(_USE_POSIX_LOCK)
#ifndef O_CLOEXEC
// kernel 2.6.23 부터 지원
#define O_CLOEXEC  0
#endif // !O_CLOEXEC
#endif // _USE_POSIX_SEMAPHORES


#define SLEEP_TIME		1000 //microsecond

workpool *workpool::m_inst = NULL;

// work
work::work()
	: m_step(-5), m_exit(false), m_enablesync("0000"), m_networkmode(false)
{
	m_nofi = NULL;
	pthread_mutex_init(&m_workmutex, NULL);
	pthread_cond_init(&m_workcond, NULL);	
#ifdef _USE_POSIX_SEMAPHORES
	m_worksem = NULL;
#elif defined (_USE_POSIX_LOCK)
	m_workfd = -1;
#endif // _USE_POSIX_SEMAPHORES
}

work::~work()
{
	pthread_mutex_destroy(&m_workmutex);
	pthread_cond_destroy(&m_workcond);
}

void* work::working(void * pdata)
{
	ostringstream msg;
	work* pObject = reinterpret_cast<work *>(pdata);

	//pthread_detach( pthread_self() );
#ifdef _DEBUG
	cout << "work::working start - " << pthread_self() << endl;
#endif // _DEBUG
	while(pObject->m_exit == false)
	{
		pthread_mutex_lock(&pObject->m_workmutex);
		pObject->m_step = -1;
		int err = pthread_cond_wait(&pObject->m_workcond, &pObject->m_workmutex);
		pObject->m_step = 0;
#ifdef _DEBUG
		cerr << "work::working - signal : " << pthread_self() << "," <<  pObject->m_exit << endl;
#endif // _DEBUG
		if ( err == 0 && pObject->m_exit == false)
		{
			pObject->m_step = 1;
			vector<string> input;
			CSyncData syncdata;
			StringSplit(pObject->m_rundata, "|", input, true);

			if (input.size() == SYNCFILE_FIELD::MAX_FILED) {

				// 서비스 간 동기화 경우 tran_id 제거된 uri 사용하므로 
				// 이 후 사용될 로직에서 편의성을 위해서 tran_id를 붙혀넣음
				if (pObject->m_type == JOB_TYPE::COPY)
					input[SYNCFILE_FIELD::uri] = "/" + input[SYNCFILE_FIELD::slave_tran_id] + input[SYNCFILE_FIELD::uri];

				syncdata.push_back(input);

				pObject->RunWork(syncdata);
			}
			else {
				msg << "validation sync file string split caution : [" << pObject->m_rundata << "]";
				
				if (pObject->m_nofi)
				{
					pObject->m_nofi(pObject->m_parm, -1);
				}
				CReportLog rlog;

				if (rlog.Init(parameter::getInstance()->getlogpath().c_str()))
				{
					rlog.Write("NOT", "[%s] [%s]", PROG_NAME, msg.str().c_str());
				}
				LOGACERR(LERR, msg);
			}
		}
		else
		{	
			pObject->m_step = -3;
			if(pObject->m_exit == false)
			{
				pObject->m_step = -4;
				msg << "Work Thread Error.";
				LOGACERR(LCRT, msg);
				if (pObject->m_nofi)
				{
					pObject->m_nofi(pObject->m_parm, -1);
				}
			}			
			else
			{
				pObject->m_step = -5;
				/* nothing -- signal exit */
			}
		}
		workpool::getInstance()->ReleaseWork(pObject);
		pObject->m_step = -6;
		pthread_mutex_unlock(&pObject->m_workmutex);
	}
#ifdef _DEBUG
	cerr << "work::working end - " << pthread_self() << endl;
#endif // _DEBUG
	return NULL;
}

bool work::init()
{
	int nRet = pthread_create(&m_thread, 0, work::working, this);
	if( nRet )
	{
		cerr << "Thread create failed.: errno: " << errno << endl;
		return false;
	}
	srand(time(NULL));
	solusleep(100);

	return true;
}

void work::setexit(bool v)
{
	m_exit = v;
	pthread_mutex_lock(&m_workmutex);
	pthread_cond_signal(&m_workcond);
	pthread_mutex_unlock(&m_workmutex);
}

void work::settargethost(vector<string>& targethost)
{
	m_vectargethost.clear();
	m_vectargethost.resize(targethost.size());
	copy( targethost.begin(), targethost.end(), m_vectargethost.begin() );
}

bool work::run(JOB_TYPE::JOB_TYPE type, string rundata, work_notifyfn nofi /* = NULL */, void * parm /* = NULL */)
{	
	m_type = type;
	m_rundata = rundata;

	m_nofi = nofi;
	m_parm = parm;
		
	ostringstream msg;
	msg << "Work Run : Sync Data - " << m_rundata;

	_LOGACOUT(LDBG, msg);

	//spin lock
	int64_t looptime = 0;
	do 
	{
		solusleep(SLEEP_TIME);
		looptime += SLEEP_TIME;
		if( looptime > (10 *1000*1000) )
		{
			msg << "Work Loop.... 10 sec over :" << m_rundata;
			_LOGACOUT(LNOT, msg);
			break;
		}
	} while (m_step != -1);
	
	pthread_cond_signal(&m_workcond);

	return true;
}

int work::RunWork(CSyncData &syncdata)
{
	int r = 0;
	short success = 0;

	ostringstream msg;

	msg << "RunWork Start : " << m_rundata;
	_LOGACOUT(LDEV2, msg);
	bool first = true;

	while (m_exit == false && !syncdata.empty())
	{
		vector<string> input = syncdata.front();
		syncdata.erase(syncdata.begin());

		string workhash;

		if (workpool::getInstance()->addworkuri(input, workhash, 180) == false)
		{
			msg << "Failed RunWork (add working uri) : " << m_rundata;
			success = -100;
			_LOGACOUT(LERR, msg);
			break;
		}

#ifdef _USE_POSIX_LOCK
		string strlockhome = "/tmp/";
		string strlock;
#endif // _USE_POSIX_LOCK
		do
		{
			// 프로세스 간 동화 목적
#ifdef _USE_POSIX_SEMAPHORES
			// lock
			if ((m_worksem = sem_open(workhash.c_str(), O_CREAT, 0777, 1)) == SEM_FAILED)
			{
				success = -101;
				sem_unlink(workhash.c_str());
				msg << "Sem Open Error. [" << m_rundata << "]";
				_LOGACOUT(LERR, msg);
				break;
			}
			sem_unlink(workhash.c_str());

			// KILL(-9) 로 프로세스 죽을 경우 block 걸림
			sem_wait(m_worksem);
#elif defined(_USE_POSIX_LOCK)
			strlock = strlockhome + workhash;
			m_workfd = open( strlock.c_str(), O_RDWR | // open the file for both read and write access 
				O_CREAT  // create file if it does not already exist 
				| O_CLOEXEC // close on execute 
				,S_IRUSR | // user permission: read 
				S_IWUSR ); // user permission: write 
			if (m_workfd < 0)
			{
				success = -102;
				msg << "lockf open fail. [" << m_rundata << "] [" << strlock << "] errno [" << errno << "]";
				_LOGACOUT(LERR, msg);
				m_workfd = -1;
				break;
			}

			// 2015.01.15 dadamin
			// EDEADLK 발생으로 중복 파일 발생하기 때문에 에러시 실패 처리함
			int l = lockf(m_workfd, F_TLOCK, 0);; // lock the "semaphore"
			if (l == -1)
			{
				success = -102;
				msg << "lockf fail. [" << m_rundata << "] [" << strlock << "] errno [" << errno << "]";
				_LOGACOUT(LERR, msg);
				break;
			}
#endif // # _USE_POSIX_SEMAPHORES

			// validation
			m_step = 3;
			CSyncData outsync;
			success = validation(first, input, outsync);
			if (success != 0)
				break;

			if (outsync.empty())
			{
				LOG(LINF, "There is no data to work with Synchronization.[%s]", m_rundata.c_str());
				break;
			}

			// diff work uri & return uri
			vector<string> verified = outsync.front();
			if (verified.empty())
			{
				success = -101;
				LOG(LCRT, "Empty Synchronization Data.[%s]", m_rundata.c_str());
				break;
			}

			string reuri = verified[SYNCFILE_FIELD::uri];

			if (input[SYNCFILE_FIELD::uri] != reuri)
			{
				if (first)
				{
					first = false;
					syncdata.clear();
					syncdata = outsync;
					LOG(LDBG, "ReWork... [%s]", m_rundata.c_str());
				}
				else
				{
					LOG(LCRT, "The other uri returned by tried again.[%s]",
						reuri.c_str());
				}
				break;
			}

			//synchronization
			m_step = 4;
			success = synchronization(verified);
			if (success != 0)
				break;
		} while (false);

#ifdef _USE_POSIX_SEMAPHORES
		if (m_worksem)
		{
			sem_post(m_worksem);
			solusleep(SLEEP_TIME);	
			sem_close(m_worksem);
			//sem_unlink(workhash.c_str());
			m_worksem = NULL;
		}
#elif defined(_USE_POSIX_LOCK)
		if(m_workfd != -1)
		{
			if (lockf(m_workfd, F_TEST, 0) < 0) {
				//unlink(strlock.c_str());
				int eno = errno;
				if (eno == EACCES || eno == EAGAIN) {
					//lockf(m_workfd, F_ULOCK, 0);
				}
			} else {
				if (success == 0) {
					unlink(strlock.c_str());
				}
			}

			close(m_workfd);
			m_workfd = -1;
		}
#endif // _USE_POSIX_SEMAPHORES

		workpool::getInstance()->delworkuri(workhash);

		if (success != 0)
			break;
	}
	
	if (m_nofi)
	{
		m_step = 6;
#ifdef _DEBUG
		cerr << "Work Thread Notify Call." << endl;
#endif //_DEBUG
		m_nofi(m_parm, success);
	}
	
	m_step = 7;
	msg << "RunWork End : " << m_rundata;
	_LOGACOUT(LDEV2, msg);

	return r;
}

int  work::validation(bool first, vector<string> &input, CSyncData &syncdata)
{
	ostringstream msg;

	msg << "Valid Start :" << input[SYNCFILE_FIELD::uri];

	_LOGACOUT(LDBG, msg);
	CValidation validate(m_type, input);
	validate.SetFHSMount(parameter::getInstance()->getfhsmount());
	int r = validate.CheckData(first, syncdata);

	return r;
}

int  work::synchronization(vector<string>& verified)
{
	ostringstream msg;

	if (verified.size() != SYNCFILE_FIELD::MAX_FILED)
	{
		msg << "synchronization data string split caution : [" << m_rundata << "]";
		LOGACERR(LERR, msg);

		return -105;
	}

	CSynchronization sync(m_type, verified);
	sync.SetTragetHost(m_vectargethost);
	sync.SetEnable(m_enablesync.c_str());
	sync.SetNetworkMode(m_networkmode);
	sync.SetFHSMount(parameter::getInstance()->getfhsmount());

	int r = sync.Execute();

	return r;
}

// work pool
workpool::workpool()
{
	pthread_mutex_init(&m_mutex, NULL);
	pthread_mutex_init(&m_mutexuri, NULL);
	pthread_mutex_init(&m_mutextry, NULL);	
}

workpool::~workpool()
{
	pthread_mutex_destroy(&m_mutex);
	pthread_mutex_destroy(&m_mutexuri);
	pthread_mutex_destroy(&m_mutextry);	
}

workpool* workpool::getInstance()
{
	if(workpool::m_inst == NULL)
	{
		workpool::m_inst = new workpool();
	}
	return workpool::m_inst;
}

void workpool::release()
{
	if( workpool::m_inst != NULL)
	{
		m_inst->DestroyPool();
		delete workpool::m_inst;
		workpool::m_inst = NULL;
	}
}

int workpool::CreatePool( int poolcnt /* = 10 */ )
{
	// pool
	for(int i=0 ; i < poolcnt ; i++)
	{
		srand ( time(NULL) );
		work * w = new work();
		if( w->init() == false)
		{
			delete w;
			break;
		}

		//m_pool.insert(make_pair(w, workpool::NOTWORK));
		m_pool.insert(pair<work *,short>(w, workpool::NOTWORK));
	}

	if( m_pool.size() < static_cast<unsigned int>(poolcnt) )
	{
		DestroyPool();
		return -1;
	}

	return 0;
}

int workpool::DestroyPool()
{
	map<work*, short>::iterator iter;
	while(m_pool.size() > 0 )
	//for( iter = m_pool.begin(); !m_pool.empty()&& iter != m_pool.end(); iter++ )
	{
		iter =  m_pool.begin();
		if(iter->second == workpool::NOTWORK)
		{
			work *data = static_cast<work *>(iter->first);
			data->setexit(true);
			pthread_join(*data->getworkhandle(), NULL);
			m_pool.erase(iter);
			delete (work *) data;
		}

		//sleep(1);
	}
	//m_pool.clear();

	return m_pool.size();
}

work* workpool::GetWorkPool(int timeout)
{
	work * r = NULL;

	int64_t usetime = 0;
	int64_t out = timeout * 1000 * 1000;

	do
	{
		pthread_mutex_lock(&m_mutex);
		map<work*, short>::iterator iter;
		for( iter = m_pool.begin(); !m_pool.empty()&& iter != m_pool.end(); iter++ )
		{
			if(iter->second == workpool::NOTWORK)
			{
				iter->second = workpool::WORKING;
				r = iter->first;
				break;
			}
		}		

		if( r == NULL )
		{
			solusleep(SLEEP_TIME);
			if( out > 0 )
			{
				usetime += SLEEP_TIME;
				if( usetime > out)
				{
					ostringstream msg;
					msg << "Get Work Pool Timeout.";
					LOGACERR(LERR, msg);
					pthread_mutex_unlock(&m_mutex);
					break;
				}
			}
		}
		pthread_mutex_unlock(&m_mutex);
	} while (r == NULL);

	return r;
}

void workpool::ReleaseWork(work *w)
{
	//pthread_mutex_lock(&m_mutex);

	map<work*, short>::iterator iter = m_pool.find(w);
	if( iter != m_pool.end() )
	{
		if(iter->second==workpool::WORKING)
			iter->second=workpool::NOTWORK;
	}
	else
	{
		ostringstream msg;
		msg << "unknown work pool.";
		LOGACERR(LWAR, msg);
	}

	//pthread_mutex_unlock(&m_mutex);
}

void workpool::printstatus(string prefixed)
{
	int w = 0, n = 0;
	map<work*, short>::iterator iter;
	for( iter = m_pool.begin(); !m_pool.empty()&& iter != m_pool.end(); iter++ )
	{
		if(iter->second == workpool::NOTWORK)
		{
			++n;
		}
		else 
		{
			++w;
		}
	}		

	ostringstream msg;
	if( prefixed.empty() == false )
		msg << "[" << prefixed << "]";

	msg << "Work Pool - " <<"total : " << m_pool.size() << "(" << w << 
		"/" << n << ")";
	_LOGACOUT(LINF, msg);
}

void workpool::printstatusex(string prefixed)
{
	ostringstream msg;
	map<work*, short>::iterator iter;
	for( iter = m_pool.begin(); !m_pool.empty()&& iter != m_pool.end(); iter++ )
	{
		if(iter->second == workpool::NOTWORK)
		{
			msg << "Work Pool(NOTWORK) - " << iter->first->getrundata() << "," <<
				iter->first->getsetp();
		}
		else 
		{
			msg << "Work Pool(WORKING) - " << iter->first->getrundata() << "," <<
				iter->first->getsetp();
		}
		if( prefixed.empty() == false )
			msg << "[" << prefixed << "]";
		_LOGACOUT(LINF, msg);
	}		
}

bool workpool::addworkuri(vector<string>& syncdata, string &workkey, int timeout)
{
	bool r = true;
	
	string key;
	ostringstream msg;
		
	if (syncdata[SYNCFILE_FIELD::uri].empty() || syncdata[SYNCFILE_FIELD::slave_tran_id].empty() )
	{
		msg << "addworkuri uri(or thran_id) empty caution.";
		LOGACERR(LERR, msg);
		return false;
	}

	string shash;
	if (syncdata[SYNCFILE_FIELD::resource_type] == TYPE_FILE)
	{
		if ( syncdata[SYNCFILE_FIELD::filename_hash].empty())
		{
			msg << "addworkuri filename_hash empty caution.";
			LOGACERR(LERR, msg);
			CReportLog rlog;
			if (rlog.Init(parameter::getInstance()->getlogpath().c_str()))
			{
				rlog.Write("ERR", "[%s] [failed work filename_hash empty caution.[uri :%s]]", PROG_NAME, syncdata[SYNCFILE_FIELD::uri].c_str());
			}
			return false;
		}
				
		size_t found = syncdata[SYNCFILE_FIELD::filename_hash].find_last_of("/");

		if (found != string::npos)
		{
			shash = syncdata[SYNCFILE_FIELD::filename_hash].substr(found + 1);
		}
		else
		{
			msg << "addworkuri not found '/'.[uri: " << syncdata[SYNCFILE_FIELD::uri] << "]";
			LOGACOUT(LWAR, msg);
		}
	}
	
	if (shash.empty())
	{
		ostringstream t;
		size_t h = __stl_hash_string(syncdata[SYNCFILE_FIELD::uri].c_str());
		t << h;
		shash = t.str();
	}

	key = syncdata[SYNCFILE_FIELD::slave_tran_id] + "." + shash;

	int64_t out = 0, usetime = 0;
	out = timeout * 1000 * 1000;

TRY_ADD_WORK_URI:
	pthread_mutex_lock(&m_mutexuri);
	workurimap::iterator find = m_workurimap.find(key);

	if (find != m_workurimap.end()) 
	{
	
		pthread_mutex_unlock(&m_mutexuri);
		
		msg << "Work uri duplication occurred and to add work uri try again." << key;
		LOGACOUT(LDEV2, msg);
						
		// try
		solusleep(SLEEP_TIME);
		usetime += SLEEP_TIME;
					
		if(timeout > 0 && usetime > out)
		{
			r = false;
			msg << "Work adduri Timeout." << key;
			LOGACERR(LERR, msg);

			pthread_mutex_unlock(&m_mutexuri);
			return false;
		}
		goto TRY_ADD_WORK_URI;		
	}
	else
	{
		// add
		workkey = key;
		m_workurimap[key] = 1;
		//pthread_mutex_unlock(&m_mutexuri);

		msg << "Add Work. ######### " << key;
		LOGACOUT(LDBG, msg);
	}
	
	pthread_mutex_unlock(&m_mutexuri);
	return r;
}

bool workpool::delworkuri(string key)
{

	pthread_mutex_lock(&m_mutexuri);
	// deleted
	m_workurimap.erase(key);
	pthread_mutex_unlock(&m_mutexuri);

	return true;
}
