#include "CopyThread.h"

#include "Logger.h"
#include "../ftsd/FtsdSocketControl.h"
#include "ProcessConfig.h"
#include "ReportLog.h"

#include <unistd.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <time.h>


CCopyThread::CCopyThread( CRebalanceData & data
	, CDataQueue<CRebalanceData>* const pQueueUpdateStandby
	, CCopyControlThread * const pControl )
	: m_data( data ), m_pQueueUpdateStandby( pQueueUpdateStandby ), m_pControl(pControl)
{
	// 멤버변수 초기화
	m_threadHandle = 0;
	
}

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

	m_pQueueUpdateStandby = NULL;
	m_pControl = NULL;
}

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

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

	// 작업할 data 객체의 각 항목이 유효한지 check

	// 1. copy target host check
	if( m_data.m_szMoveHost.size() < 7 )
	{
		LOG( LERR, "CopyThread: copy target host[%s] not valid.", m_data.m_szMoveHost.c_str() );
		return false;
	}

	// 2. origin host & origin content check
	if( m_data.m_szOriginHost.size() < 7 || m_data.m_szOriginFileNameHash.size() < 5 )
	{
		LOG( LERR, "CopyThread: copy origin host[%s], filename[%s] not valid."
			, m_data.m_szOriginHost.c_str(), m_data.m_szOriginFileNameHash.c_str() );
		return false;
	}

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

		return false;
	}

	return true;
}

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

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

	pthread_testcancel();

	// Thread 동작시 control class 에 count 증가 처리
	pObject->m_pControl->PlusCopyThreadCount();

	// 실제 작업 수행
	pObject->Execute();
	
	// Thread 동작 완료시 control class 에 count 감소 처리
	pObject->m_pControl->MinusCopyThreadCount();

	pthread_testcancel();

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

	// 객체 자동 delete 처리
	delete pObject;
	return NULL;
}

void CCopyThread::Execute(  )
{
	// FHS ftsd 와 통신 처리 담당한 socket 객체 생성
	CFtsdSocketControl fhs;

	// origin FHS connect
	bool bResult 
		= fhs.ConnectTarget( m_data.m_szOriginHost, CProcessConfig::GetInstance()->GetFhsTransferDaemonPort() );

	if( bResult == false )
	{
		// origin fhs connect fail.
		LOG( LERR, "CopyThread: origin fhs[%s][%u] connect fail. [%s]"
			, m_data.m_szOriginHost.c_str(), CProcessConfig::GetInstance()->GetFhsTransferDaemonPort()
			, m_data.m_szURI.c_str() );

		return;
	}

	// target FHS 정보를 vector 로 저장 처리
	std::vector<std::string> vecTargetFHS;
	vecTargetFHS.clear();
	vecTargetFHS.push_back( m_data.m_szMoveHost );


	// 연결된 FHS 로 content copy 처리 요청 전달.
	bResult = fhs.SendFileReplicationRequest( m_data.m_szOriginFileNameHash
											, m_data.m_uOriginContentLength
											, CProcessConfig::GetInstance()->GetHashCheckLevel()
											, true
											, vecTargetFHS );

	if( bResult == false )
	{
		// content copy 요청 실패
		LOG( LERR, "CopyThread: content copy request send fail to origin fhs.[%s][%s]"
			, m_data.m_szOriginHost.c_str(), m_data.m_szURI.c_str() );

		return;
	}

	// FHS 로 부터 처리결과 수신 대기

	// response timeout 처리 관련 변수
	struct timespec timeStart, timeNow;
	double timeElapsed = 0.0;
	memset( &timeStart, 0x00, sizeof( struct timespec ) );
	memset( &timeNow, 0x00, sizeof( struct timespec ) );
	clock_gettime( CLOCK_MONOTONIC, &timeStart );


	// 응답 결과 저장을 위한 변수
	int nResult;
	bool bSuccess = false;
	std::string szErrorMessage;
	std::map<std::string, std::string> mapSuccessFHS;
	std::map<std::string, std::string> mapFailFHS;

	while( 1 )
	{
		nResult = fhs.GetFileReplicationResult( 3, bSuccess, szErrorMessage, mapSuccessFHS, mapFailFHS );

		if( nResult == 0 )
		{
			// 지정된 시간동안 결과 수신 못한 경우
			if( fhs.SendAliveCheck() == false )
			{
				// 다음  loop 에서 -1 연결 종료 처리되므로.. 여기서는 로깅만..
				LOG( LERR, "CopyThread: content copy alive check fail.[%s][%s]"
					, m_data.m_szOriginHost.c_str(), m_data.m_szURI.c_str() );

			}
			else
			{
				// timeout 발생했으므로.. 2 sec 대기 후 다시 시도 
				sleep( 2 );
			}
		}
		else if( nResult == 1 )
		{
			// 처리 시간 계산
			clock_gettime( CLOCK_MONOTONIC, &timeNow );
			timeElapsed = (double)((timeNow.tv_sec - timeStart.tv_sec) * 1.0e9 + (timeNow.tv_nsec - timeStart.tv_nsec)) / 1.0e9;


			// FHS 로 부터 응답 수신
			if( bSuccess == true )
			{
				// Copy 처리 결과가.. mapSuccessFHS 에 저장된 상태면.. 성공
				// mapFailFHS 에 저장된 상태면 뭔가 오류가 발생한 상황으로 판단해야 함
				std::map<std::string, std::string>::iterator itFHS;

				if( mapSuccessFHS.empty() == false )
				{
					// 성공 FHS 목록에 존재하는 경우...  정상 copy 처리된 것으로 판단 
					itFHS = mapSuccessFHS.begin();

					m_data.m_uMoveResult = MOVE_OK;
					m_data.m_szMoveHost = itFHS->first;
					m_data.m_szMoveFileNameHash = itFHS->second;
					m_data.m_uMoveContentLength = m_data.m_uOriginContentLength;

					// RCDB update queue 에 PUSH 처리
					if( m_pQueueUpdateStandby->Push( m_data ) == false )
					{
						// 실패시 3 sec sleep 후 1회만 재시도.
						sleep( 3 );
						if( m_pQueueUpdateStandby->Push( m_data ) == false )
						{
							LOG( LERR, "CopyThread: copy ok [%.2lf sec]. but update queue push fail. OP manual delete need.[%s %s]"
								, timeElapsed
								, m_data.m_szMoveHost.c_str(), m_data.m_szMoveFileNameHash.c_str() );
							break;
						}
					}

					// 관련 내역 로깅
					_LOG( LINF, "CopyThread[%lu]: copy ok. origin [%s svc:%s id:%llu size:%llu uri:%s]"
						, pthread_self()
						, m_data.m_szOriginHost.c_str(), m_data.m_szServiceSeq.c_str()
						, m_data.m_uResourceId,  m_data.m_uOriginContentLength
						, m_data.m_szURI.c_str() );
					_LOG( LINF, "CopyThread[%lu]: copy ok [%.2lf sec]. target [%s size:%llu %s]"
						, pthread_self()
						, timeElapsed
						, m_data.m_szMoveHost.c_str(), m_data.m_uMoveContentLength, m_data.m_szMoveFileNameHash.c_str() );
				}
				else
				{
					// fail 목록에 포함된 상태.
					if( mapFailFHS.empty() == false )
					{
						itFHS = mapFailFHS.begin();

						m_data.m_uMoveResult = MOVE_ERROR_CREATE;
						m_data.m_szMoveHost = itFHS->first;
						szErrorMessage = itFHS->second;

						// 실패했으므로.. RCDB update queue 에 push 하지 않고...
						// 로깅만 하고 종료한다.
						LOG( LERR, "CopyThread: copy fail [%.2lf sec]. [%s svc:%s id:%llu size:%llu]->[%s][%s]"
							, timeElapsed
							, m_data.m_szOriginHost.c_str(), m_data.m_szServiceSeq.c_str()
							, m_data.m_uResourceId, m_data.m_uOriginContentLength
							, m_data.m_szMoveHost.c_str(), szErrorMessage.c_str() 
						);
					}
					else
					{
						// success, fail map 이 모두 빈 상태.. ?
						// 이런 상황은 발생하면 안되는데...
						LOG( LERR, "CopyThread: copy fail [%.2lf sec]. success-fail map all empty.[%s svc:%s id:%llu size:%llu]->[%s]"
							, timeElapsed
							, m_data.m_szOriginHost.c_str(), m_data.m_szServiceSeq.c_str()
							, m_data.m_uResourceId, m_data.m_uOriginContentLength
							, m_data.m_szMoveHost.c_str() 
						);
					}
				}
			}
			else
			{
				// content 이상으로 copy 처리하지 못한 경우.
				LOG( LERR, "CopyThread: copy fail [%.2lf sec]. check content [%s svc:%s id:%llu size:%llu]->[%s][%s]"
					, timeElapsed
					, m_data.m_szOriginHost.c_str(), m_data.m_szServiceSeq.c_str()
					, m_data.m_uResourceId, m_data.m_uOriginContentLength
					, m_data.m_szMoveHost.c_str(), szErrorMessage.c_str()
				);

				// source file access 오류인 경우..
				// - report 로깅 
				if( strncmp( szErrorMessage.c_str(), "source file access fail", 23 ) == 0 )
				{
					CReportLog reportLog;
					std::string errorMessage;
					errorMessage.clear();

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

			// 처리 결과 수신에 따른 loop 종료
			break;

		}
		else if( nResult == -1 )
		{
			// 처리 시간 계산
			clock_gettime( CLOCK_MONOTONIC, &timeNow ); 
			timeElapsed = (double)((timeNow.tv_sec - timeStart.tv_sec) * 1.0e9 + (timeNow.tv_nsec - timeStart.tv_nsec)) / 1.0e9;

			// FHS 와 연결 끝어진 경우...
			LOG( LERR, "CopyThread: copy fail [%.2lf sec]. fhs connection broken.[%s svc:%s id:%llu size:%llu]->[%s]"
				, timeElapsed, m_data.m_szOriginHost.c_str(), m_data.m_szServiceSeq.c_str()
				, m_data.m_uResourceId, m_data.m_uOriginContentLength
				, m_data.m_szMoveHost.c_str() );

			break;
		}
		else if( nResult == 2 )
		{
			// Replication 응답이 아닌 경우.. 무시
			;
		}
		else
		{
			// 미정의된 상황
			LOG( LERR, "CopyThread: receive undefine result.[%d] [%s][%s][%llu]->[%s]"
				, nResult
				, m_data.m_szURI.c_str(), m_data.m_szOriginHost.c_str(), m_data.m_uOriginContentLength
				, m_data.m_szMoveHost.c_str() );
			break;
		}

		// Total Reponse 대기 시간 확인
		clock_gettime( CLOCK_MONOTONIC, &timeNow );
		timeElapsed = (double)((timeNow.tv_sec - timeStart.tv_sec) * 1.0e9 + (timeNow.tv_nsec - timeStart.tv_nsec)) / 1.0e9;
		if( timeElapsed > 1200.0 )	// 20분 이상 대기하는 경우 - 장비별 재선택 제한시간이 20 분 정도이므로.. 
		{
			LOG( LERR, "CopyThread: copy fail. response wait %.2lf sec, but timeout 1200 sec.[%s svc:%s id:%llu size:%llu]->[%s]"
				, timeElapsed
				, m_data.m_szOriginHost.c_str(), m_data.m_szServiceSeq.c_str()
				, m_data.m_uResourceId, m_data.m_uOriginContentLength
				, m_data.m_szMoveHost.c_str() );

			break;
		}

	} // while(1)

}

