interactive/rcts/rc_syncd/src/WorkPool.cpp

/***************************************************************************
                Work Pool
                -----------------------------------------
	begin      : 2011/10/27
	copyright  : (C) 2011 SolutionBox Inc.
	author     : Service 1 Team
	email      : svc1@solbox.com
	version    : 3.0.1

    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 "WorkPool.h"
#include "Validation.h"
#include "Logger.h"
#include "Util.h"

#include <unistd.h>

#include <errno.h>
#include <iostream>

#define	DIFF_SPLIT_CNT	6		// diff file format split count
#define SLEEP_TIME		1000	//microsecond

#define  _LOGMSG(level, msg) \
	_LOG(level, "%s", msg.str().c_str());\
	msg.str("");

CWorkPool *CWorkPool::m_inst = NULL;

// work
CWork::CWork()
	: m_step(-5), m_exit(false), m_enablesync("0000")
{
	m_nofi = NULL;
	pthread_mutex_init(&m_workmutex, NULL);
	pthread_cond_init(&m_workcond, NULL);	
}

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

void* CWork::working(void * pdata)
{
	ostringstream msg;
	CWork* pObject = reinterpret_cast<CWork *>(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;
			short success = 0;
						
			msg << "Valid Start : " << pObject->m_rundata;
			_LOGMSG(LDBG, msg);

			do
			{
				// add uri 
				pObject->m_step = 2;
				if (CWorkPool::getInstance()->addworkuri(pObject->m_syncinfo.sync_master,
														pObject->m_rundata, 
														pObject->m_workuri, true)
					== false)
				{
					LOG(LERR, "Failed work uri map add [%s].", pObject->m_rundata.c_str());
					success = -1;
					break;
				}

				// validation	
				pObject->m_step = 3;
				CValidation v(pObject->m_syncinfo, pObject->m_rundata);

				if (v.CheckValidation() == false)
					success = -2;

				pObject->m_step = 4;
			} while (false);

			if(pObject->m_nofi)
			{
				pObject->m_step = 6;
#ifdef _DEBUG
				cerr << "Work Thread Notify Call."<< endl;
#endif //_DEBUG
				pObject->m_nofi(pObject->m_parm, success);
			}
			pObject->m_step = 7;
			
			msg << "Valid End : " << pObject->m_rundata;
			_LOGMSG(LDBG, msg);
			
			if(pObject->m_workuri.size())
			{
				CWorkPool::getInstance()->delworkuri(pObject->m_workuri);
			}
			CWorkPool::getInstance()->ReleaseWork(pObject);
		}
		else
		{	
			pObject->m_step = -3;
			if(pObject->m_exit == false)
			{
				pObject->m_step = -4;
				msg << "Work Thread Error.";
				//LOGACERR(LCRT, msg);
				LOG(LCRT, msg.str().c_str());
			}			
			else
			{
				pObject->m_step = -5;
				/* nothing -- signal exit */
			}
		}
		pObject->m_step = -6;
		pthread_mutex_unlock(&pObject->m_workmutex);
	}
#ifdef _DEBUG
	cerr << "work::working end - " << pthread_self() << endl;
#endif // _DEBUG
	return NULL;
}

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

	sleep(0);

	return true;
}

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

bool CWork::run(CSyncInfo info, string rundata, work_notifyfn nofi /* = NULL */, void * parm /* = NULL */)
{	
	m_syncinfo = info;
	m_rundata = rundata;
	
	m_parm = parm;
	m_nofi = nofi;
	
	ostringstream msg;
	//msg << "Work Run : TranID(" <<m_master.m_tranid <<"=>" << m_slave.m_tranid << 
	//	"), UserSeq(" << m_master.m_userseq << "=>" << m_slave.m_userseq << ")" <<
	//	", 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;
			LOG(LNOT, "%s", msg.str().c_str());
			break;
		}
	} while (m_step != -1);
	
	pthread_cond_signal(&m_workcond);

	return true;
}

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

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

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

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

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

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

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

	return 0;
}

int CWorkPool::DestroyPool()
{
	map<CWork*, 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 == CWorkPool::NOTWORK)
		{
			CWork *data = static_cast<CWork *>(iter->first);
			data->setexit(true);
			pthread_join(*data->getworkhandle(), NULL);
			m_pool.erase(iter);
			delete (CWork *)data;
		}

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

	return m_pool.size();
}

CWork* CWorkPool::GetWorkPool(int timeout)
{
	CWork * r = NULL;

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

	do
	{
		pthread_mutex_lock(&m_mutex);
		map<CWork*, short>::iterator iter;
		for( iter = m_pool.begin(); !m_pool.empty()&& iter != m_pool.end(); iter++ )
		{
			if(iter->second == CWorkPool::NOTWORK)
			{
				iter->second = CWorkPool::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);
					LOG(LERR, msg.str().c_str());
					pthread_mutex_unlock(&m_mutex);
					break;
				}
			}
		}
		pthread_mutex_unlock(&m_mutex);
	} while (r == NULL);

	return r;
}

void CWorkPool::ReleaseWork(CWork *w)
{
	//pthread_mutex_lock(&m_mutex);

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

	//pthread_mutex_unlock(&m_mutex);
}

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

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

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

void CWorkPool::printstatusex(string prefixed)
{
	ostringstream msg;
	map<CWork*, short>::iterator iter;
	for( iter = m_pool.begin(); !m_pool.empty()&& iter != m_pool.end(); iter++ )
	{
		if(iter->second == CWorkPool::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);
		_LOG(LINF, "%s", msg.str().c_str());
	}		
}

bool CWorkPool::addworkuri(string tranid, string uri, string &workuri, 
					bool sourcedata, int timeout)
{
	bool r = true;
	
	string key;
	ostringstream msg;
	
	if(sourcedata)
	{
		vector<string> strSplit;
		StringSplit(uri, "|", strSplit, true);
		if(strSplit.size() != DIFF_SPLIT_CNT)
		{
			pthread_mutex_unlock(&m_mutexuri);
			msg << "addworkuri sync file string split caution : [" << uri << "]";
			//LOGACERR(LERR, msg);
			LOG(LERR, "%s", msg.str().c_str());
			return false;
		}
		key	= "/" + tranid + strSplit[0] + "|[Size=" + strSplit[2] + "]";
	}
	else 
		key = "/"+ tranid + uri;

	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;
		_LOGMSG(LDBG, msg);
						
		// try
		solusleep(SLEEP_TIME);
		usetime += SLEEP_TIME;
					
		if(timeout > 0 && usetime > out)
		{
			r = false;
			msg << "Work adduri Timeout." << key;
			//LOGACERR(LERR, msg);
			LOG(LERR, "%s", msg.str().c_str());

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

bool CWorkPool::delworkuri(string key)
{
	pthread_mutex_lock(&m_mutexuri);
	// deleted
	m_workurimap.erase(key);
	pthread_mutex_unlock(&m_mutexuri);
	return true;
}