#include "CopyControlThread.h"

#include "Logger.h"
#include "CopyThread.h"
#include "ReportLog.h"
#include "ProcessConfig.h"

#include <unistd.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <list>


// 매 Loop 마다 Sleep 시간 정보 정의 ( sec 단위)
#define SLEEP_TIME_DEFAULT		 5 
#define SLEEP_TIME_MAX			10	


// 본 값은 Makefile 에서 설정 
//#define MAX_COPY_THREAD_COUNT		2


CCopyControlThread::CCopyControlThread( CDataQueue<CRebalanceData> * const pQueueMoveStandby
	, CDataQueue<CRebalanceData> * const pQueueUpdateStandby
	, CShmInfoThread * const pShmInfoThread )
	: m_pQueueMoveStandby( pQueueMoveStandby )
	, m_pQueueUpdateStandby( pQueueUpdateStandby )
	, m_pShmInfoThread( pShmInfoThread )
{
	// 멤버변수 초기화

	m_threadHandle = 0;		// thread handle 초기화

	m_uSleep = SLEEP_TIME_DEFAULT;

	// Copy Thread count 관련 변수 및 lock 객체 초기화
	m_nCurrentCopyThreadCount = 0;
	pthread_mutex_init( &m_mutexCopyThreadCount, NULL );

	m_uTargetFhsSelectFailCount = 0;
}


CCopyControlThread::~CCopyControlThread()
{
	// thread 가 동작 중인 경우.. thread 중지 처리
	if( m_threadHandle != 0 )
	{
		pthread_cancel( m_threadHandle );
	}

	sleep( 1 );	// thread 종료 잠시 대기

	pthread_mutex_destroy( &m_mutexCopyThreadCount );


	// 혹시나 해서 정석대로 NULL 처리 추가함.
	if( m_pQueueMoveStandby != NULL )
		m_pQueueMoveStandby = NULL;
	if( m_pQueueUpdateStandby != NULL )
		m_pQueueUpdateStandby = NULL;
	if( m_pShmInfoThread != NULL )
		m_pShmInfoThread = NULL;
}


///////////////////////////////////////////////////

bool CCopyControlThread::Start()
{
	// 멤버 변수 유효성 check
	if( m_pQueueMoveStandby == NULL || m_pQueueUpdateStandby == NULL )
	{
		LOG( LERR, "CCopyControlThread: queue object pointer not valid." );
		return false;
	}

	// RCDB 접속 정보 load
	m_rcdbInfo.Load();


	// 작업 thread 생성
	int nResult = pthread_create( &m_threadHandle, NULL, CCopyControlThread::threadFunc, this );
	if( nResult != 0 )
	{
		// thread 생성 실패
		int errorNum = errno;
		LOG( LERR, "CCopyControlThread: Thread create failed.[%d][%s]", errorNum, strerror( errorNum ) );

		return false;
	}

	return true;
}

void CCopyControlThread::PlusCopyThreadCount()
{
	pthread_mutex_lock( &m_mutexCopyThreadCount );
	++m_nCurrentCopyThreadCount;
	pthread_mutex_unlock( &m_mutexCopyThreadCount );

	return;
}

void CCopyControlThread::MinusCopyThreadCount()
{
	pthread_mutex_lock( &m_mutexCopyThreadCount );
	--m_nCurrentCopyThreadCount;
	pthread_mutex_unlock( &m_mutexCopyThreadCount );

	return;
}

int CCopyControlThread::GetCopyThreadCount()
{
	return m_nCurrentCopyThreadCount;
}

int CCopyControlThread::GetMaxCopyThreadCount()
{
	return MAX_COPY_THREAD_COUNT;
}

///////////////////////////////////////////////////

void * CCopyControlThread::threadFunc( void * arg )
{
	CCopyControlThread * pObject = reinterpret_cast<CCopyControlThread *>(arg);
	pthread_detach( pthread_self() );

	while( 1 )
	{
		pthread_testcancel();
		pObject->Execute();
		pthread_testcancel();

		// 지정된 주기로 실행
		sleep( pObject->m_uSleep );
	}

	// thread 종료시 handle 정보 초기화
	pObject->m_threadHandle = 0;

	return NULL;
}

void CCopyControlThread::Execute()
{
	// sleep time 조정 처리
	m_uSleep = SLEEP_TIME_DEFAULT;

	// COPY 처리할 대상 content 없다면...
	if( m_pQueueMoveStandby->GetSize() == 0 )
	{
		// MAX sleep 처리
		m_uSleep = SLEEP_TIME_MAX;

		_LOG( LDBG, "[COPY   ] standby queue empty. max sleep %u sec.", m_uSleep );
		return;
	}
	else
	{
		// RCDB update standby queue 에 여유 공간이 있는지 check
		// - 없는 경우 RCDB 장애 등의 상황이므로... MAX sleep 처리한다.
		unsigned int uUpdatePushEnableCount = m_pQueueUpdateStandby->GetMaxSize() - m_pQueueUpdateStandby->GetSize();

		// RCDB update standby queue 에 10% 정도 여유 공간이 없으면...
		if( uUpdatePushEnableCount < (m_pQueueUpdateStandby->GetMaxSize() /10) )
		{
			m_uSleep = SLEEP_TIME_MAX; 

			_LOG( LDBG, "[COPY   ] rcdb update queue not enough (%u %u). max sleep %u sec."
				, uUpdatePushEnableCount, m_pQueueUpdateStandby->GetMaxSize() / 10,  m_uSleep );
			return;
		}
	}

	// 생성 가능한 copy thread 가 존재하는지 확인
	unsigned int uEnableThreadCount = GetCopyThreadEnableCount();

	// 생성 가능한 copy therad 가 없는 경우... 잠시 대기 후 다시 시도
	if( uEnableThreadCount == 0 )
	{
		m_uSleep = SLEEP_TIME_DEFAULT;

		_LOG( LDBG, "[COPY   ] copy thread all working [%u / %u]. sleep %u sec."
			, uEnableThreadCount, MAX_COPY_THREAD_COUNT, m_uSleep );
		return;
	}

	// ShmInfoThread 객체에서 COPY 가능 FHS 정보가 유효한지 check
	// - COPY 가능 FHS 가 없는 경우.... 잠시 대기후 다시 시도
	if( m_pShmInfoThread->IsValidCopyTargetFhs() == false )
	{
		m_uSleep = SLEEP_TIME_MAX;

		LOG( LERR, "CopyControlThread: valid copy target FHS not exist. sleep %u sec.", m_uSleep );
		return;
	}

	// RCDB 처리를 수행할 객체 생성
	// RCDB 와 연결은 필요시마다 연결 후 해제 처리한다. (세션 유지 않도록...)		
	DataBase rcdb;

	// RCDB 와 연결을 시도한다.
	if( rcdb.PgOpenDB( m_rcdbInfo.m_strRcdbIp, m_rcdbInfo.m_nRcdbPort, m_rcdbInfo.m_strRcdbName
		, m_rcdbInfo.m_strRcdbAcct, m_rcdbInfo.m_strRcdbAcctPw ) == NULL )
	{
		// 연결 실패시...
		m_uSleep = SLEEP_TIME_DEFAULT;
		LOG( LERR, "CopyControlThread: RCDB connection failed. sleep %u sec.", m_uSleep );
		return;
	}

	// move standby queue 에서 생성 가능한 copy theread 갯수만큼 pop 처리한다.
	CRebalanceData data;
	std::list<CRebalanceData> listWork;

	while( uEnableThreadCount > listWork.size() && m_pQueueMoveStandby->GetSize() > 0 )
	{
		// Move 대기 queue 에서 data 가져와서 pop 처리하고
		data = m_pQueueMoveStandby->Front();
		m_pQueueMoveStandby->Pop();

		// 작업 list 에 등록
		listWork.push_back( data );
	}

	_LOG( LINF, "[COPY   ] start." );

	// list 에 등록된 각 content 에 대해 해당 content 를 소유한 FHS 정보를 조회한 후 이를 처리할 copy thread 를 생성 처리한다.
	int nResult = 0;
	std::list<std::string> listIgnoreFHS;

	while( listWork.empty() == false )
	{
		data = listWork.front();
		listWork.pop_front();
		
		// RCDB 조회를 통하여 FHS 정보 조회
		listIgnoreFHS.clear();
		if( GetMetaHostList( data, rcdb, listIgnoreFHS ) == false )
		{
			// meta 정보 조회 실패 또는 해당 content 가 삭제된 상황인 경우...
			// 해당 함수에서 로깅 했으므로... flow 만 처리
			// 다시 대기 queue 에 넣을 수 없으므로 그냥 버린다.
			LOG( LERR, "[COPY   ] ignore fhs check fail. throw. [%s svc:%s id:%llu size:%llu]"
				, data.m_szOriginHost.c_str(), data.m_szServiceSeq.c_str()
				, data.m_uResourceId, data.m_uOriginContentLength );

			continue;
		}

		// COPY 처리할 FHS 선정
		std::string szTargetHost;

		nResult = m_pShmInfoThread->GetCopyTargetFhs( szTargetHost, listIgnoreFHS );
		if( nResult == 1 )			// COPY 대상 FHS 가 선정된 경우
		{
			data.m_szMoveHost = szTargetHost;
		}
		else if( nResult == -1 )	// 복사 가능 물리적 FHS 가 없는 경우 
		{
			// 다시 대기 queue 에 넣을 수 없으므로. 버린다.
			LOG( LWAR, "[COPY   ] proper copy target FHS not exist. throw. [%s svc:%s id:%llu size:%llu]"
				, data.m_szOriginHost.c_str(), data.m_szServiceSeq.c_str()
				, data.m_uResourceId, data.m_uOriginContentLength );

			// 다음 loop 는 좀 더 지연 후 동작하도록 한다.
			m_uSleep = SLEEP_TIME_MAX;
			continue;
		}
		else	
		{
			// 복사 가능 물리 FHS 는 존재하지만 ignore 에 포함되어 실제로는 가능 FHS 가 없는 경우
			// - 다시 대기 queue 에 넣을 수 없으므로. 버린다.
			LOG( LWAR, "[COPY   ] available copy target FHS not exist by ignore list. throw. [%s svc:%s id:%llu size:%llu]"
				, data.m_szOriginHost.c_str(), data.m_szServiceSeq.c_str()
				, data.m_uResourceId, data.m_uOriginContentLength );

			// 2021-01-07 NEW huibong 이동 대상 content 선택 관련 mode 기능 추가 (#33306)
			// - copy 작업 관련 ignore list 에 의한 FHS 선정 실패가 지정된 횟수를 초과되는 경우.. 
			// - content select mode (ContentSelectMode) 를 big size 순서-> RANDOM 모드로 변경 처리 코드 추가

			// Copy 대상 FHS 선택관련 ignore list 에 의한 실패 count 증가 처리
			m_uTargetFhsSelectFailCount++;

			// 만약 실패 값이 지정된 값 이상인 경우 .. content SELECT mode 를 random mode 로 변경 처리한다.
			if( m_uTargetFhsSelectFailCount > TARGET_FHS_SELECT_MAX_FAIL )
			{
				if( CProcessConfig::GetInstance()->GetContentSelectMode() != SELECT_MODE_RANDOM )
				{
					CProcessConfig::GetInstance()->SetContentSelectMode( SELECT_MODE_RANDOM );
			
					LOG( LWAR, "[COPY   ] content select mode change to RANDOM by copy faiil count over [%u/%d]"
						, m_uTargetFhsSelectFailCount, TARGET_FHS_SELECT_MAX_FAIL );
				}

				// 실패 횟수 count 변수 초기화
				m_uTargetFhsSelectFailCount = 0;
			}

			continue;
		}
		

		// Copy 처리를 수행할 Thread 생성 처리
		CCopyThread * pCopyThread = new CCopyThread( data, m_pQueueUpdateStandby, this );

		// Thread 생성 또는 start 처리 실패시
		// - thread 생성이 실패할 가능성은 거의 없고..
		// - 아마 CopyThread 객체의 Start 처리시 인자 check 등에서 오류일 가능성이 높으므로.. 
		// - 재시도 하지 않고.. 그냥 skip 처리한다.
		if( pCopyThread == NULL || pCopyThread->Start() == false )
		{
			// thread 정리
			if( pCopyThread != NULL )
			{
				delete pCopyThread;
				pCopyThread = NULL;
			}

			// 실패 내역 로깅
			LOG( LWAR, "[COPY   ] copy thread create or start fail. fhs[%s] svc[%s] resouce[%llu]."
				, data.m_szOriginHost.c_str(), data.m_szServiceSeq.c_str(), data.m_uResourceId );
		}
	}

	// RCDB 와의 연결을 명시적으로 해제 처리
	rcdb.PgCloseDB();

	return;
}


unsigned int CCopyControlThread::GetCopyThreadEnableCount( )
{
	unsigned int uEnableCount = 0;

	// lock 설정 후 추가 생성 가능한 Copy thread 계산한다.
	pthread_mutex_lock( &m_mutexCopyThreadCount );
	uEnableCount = MAX_COPY_THREAD_COUNT - m_nCurrentCopyThreadCount;
	pthread_mutex_unlock( &m_mutexCopyThreadCount );

	// 값 보정
	if( uEnableCount < 0 )
		uEnableCount = 0;

	return uEnableCount;
}


bool CCopyControlThread::GetMetaHostList( CRebalanceData & data, DataBase & rcdb, std::list<std::string> & listHost )
{
	// RCDB 연결은 이미 되어 있는 상태

	char szQuery[4096];
	int nResult = 0;

	// 해당 content 를 가진 모든 장비를 조회
	// - 원본, 복제본은 상관 없으며... cache 는 같은 장비에 있어도 되지만.. 혹시나 해서 제외 목록에 포함
	// - 삭제된 content 역시 rc_gcmd 가 작업 중 삭제시킬 수 있으므로.. 제외.
	// - uri 정보는 query 에 적합하게 변환해서 query 수행.
	// - 원본에 정상적으로 동작하는지 check 한 후 존재할 경우.. 장비 목록을 추출.
	snprintf( szQuery, 4096 - 1
		, "SELECT host_name FROM t_meta_%s WHERE uri = "
		  "( SELECT uri FROM t_meta_%s WHERE uri='%s' AND deleted_yn = 'N' AND is_cache = 'N' "
		  "  AND resource_type = 0 AND get_content_length != 0); "
		, data.m_szServiceSeq.c_str()
		, data.m_szServiceSeq.c_str()
		, data.GetQueryURI().c_str() );

	rcdb.PgDoExec( szQuery );

	nResult = rcdb.PgResult( DataBase::NOT_CLEAR );
	if( nResult < 0 )
	{
		// Query 수행 실패시
		LOG( LERR, "[COPY   ] ignore fhs meta select error. [%d][%s][%s]"
			, nResult, rcdb.GetErrorMessage().c_str(), szQuery );

		// 중복 URI 가 존재하는 경우...
		// nResult = -4, Error message [ERROR:  more than one row returned by a subquery used as an expression] 발생
		// 해당 상황인 경우.. 운영자가 조치할 수 있도록  report 로깅 처리
		if( nResult == -4 
			&& strncmp( rcdb.GetErrorMessage().c_str(), "ERROR:  more than one row returned", 34) == 0 )
		{
			CReportLog reportLog;
			std::string strErrorMessage;
			strErrorMessage.clear();

			if( reportLog.Init( CProcessConfig::GetInstance()->GetLogPath() ) == false )
			{
				reportLog.GetLastError( strErrorMessage );
				LOG( LERR, "[COPY   ] report log init failed.[%s]", strErrorMessage.c_str() );
			}
			else
			{
				reportLog.Write( "ERR", "[%s] [duplicate content. please check and delete. host:%s svc:%s id:%llu uri:'%s' ]"
					, PROG_NAME
					, data.m_szOriginHost.c_str(), data.m_szServiceSeq.c_str(), data.m_uResourceId
					, data.GetQueryURI().c_str() );

				LOG( LWAR, "[COPY   ] duplicate content found. [%s svc:%s id:%llu uri:%s][%s]"
					, data.m_szOriginHost.c_str(), data.m_szServiceSeq.c_str(), data.m_uResourceId
					, data.GetQueryURI().c_str(), rcdb.GetErrorMessage().c_str() );
			}
		}

		// 다음 Query 수행을 위해 조회 결과 set 을 clear 처리
		rcdb.PgClear();

		return false;
	}

	// 결과 Row 수 확인 
	nResult = rcdb.GetNoTuples();
	if( nResult < 0 )
	{
		// 오류 발생시
		LOG( LERR, "[COPY   ] copy meta ignore fhs select tuple error. [%d][%s][%s]"
			, nResult, rcdb.GetErrorMessage().c_str(), szQuery );

		rcdb.PgClear();
		return false;
	}
	else if( nResult == 0 )
	{
		// 조회 결과가 없는 경우.
		// Copy 대상 content 가 삭제된 경우....
		LOG( LERR, "[COPY   ] copy meta ignore fhs select not found. [%d][%s]", nResult, szQuery );

		// 다음 Query 수행을 위해 조회 결과 set 을 clear 처리
		rcdb.PgClear();
		return false;
	}
	else
	{
		// 조회 결과가 존재.
		std::string szTemp;
		listHost.clear();

		// 조회 목록을 list add.
		for( int i = 0; i < nResult; i++ )
		{
			szTemp = rcdb.GetValue( i, 0 );
			listHost.push_back( szTemp );
		}
	}

	// 다음 Query 수행을 위해 조회 결과 set 을 clear 처리
	rcdb.PgClear();

	// 처리 결과 로깅
	_LOG( LDBG, "[COPY   ] copy meta ignore FHS select ok. [%d][%s]", nResult, szQuery );
	
	return true;
}

