#include "Worker.h"

#include "ProcessRename.h"
#include "ServiceConfig.h"
#include "Logger.h"
#include "DataQueue.hpp"
#include "ReplicationData.h"
#include "CacheRequestData.h"
#include "CacheData.h"
#include "ProcessStatus.h"
#include "ReplicateSelect.h"
#include "ReplicateInsert.h"
#include "ProperFhs.h"
#include "ReplicateControl.h"
#include "FimngdSocketControl.h"
#include "CacheJobList.h"
#include "CacheDelete.h"
#include "CacheInsert.h"
#include "CacheOrder.h"
#include "CacheControl.h"

#include <unistd.h>
#include <signal.h>
#include <netinet/in.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <time.h>


// hot content 정보 관련 fimngd 로 부터 TCP 접속 요청 수신시
// 이를 처리할 Thread 에게 전달할 정보 구조체
struct stFhsReceiveThreadArg {

	// client socket descriptor
	int clientSocketDesc;

	// Client 의 IP 주소 정보를 저장한 변수
	char szClientAddress[INET_ADDRSTRLEN];

	// 요청된 Cache 정보를 저장하기 위한 Queue 에 대한 포인터.
	CDataQueue<CCacheRequestData> * pQueueCacheRequest;
};


/// @brief Worker Process 의 main 함수
int WorkerMain( int socketListenHotContent )
{
	// Process Rename
	set_ps_display(PROG_NAME": Worker [initialize process]", false);

	// Set Signal Handler
	SetSignalWorker();
    
	// 로그 출력 순서가 Main 프로세스와 꼬여서... 1초간 대기 후 진행
	sleep( 1 );
	


	// 1. 각 모듈 생성 및 초기화.
	//    - 초기화 실패시 EXIT_FAILURE 를 return 하여 프로세스가 종료되도록 한다.
	
	// 1.1 Data 처리를 위한 객체 생성
	
	// 복제대상 정보를 관리하는 큐.
	CDataQueue<CSourceData> queueReplicate;
	queueReplicate.SetMaxSize( CServiceConfig::GetInstance()->GetMaxQueueSize() );

	// 복제 시도 후 실패한 정보를 관리하는 큐.
	CDataQueue<CRetryData> queueRetry;
	queueRetry.SetMaxSize( CServiceConfig::GetInstance()->GetMaxQueueSize() );

	// 복제 완료된 결과를 관리 하는 큐.
	CDataQueue<CCompleteData> queueComplete;
	queueComplete.SetMaxSize( CServiceConfig::GetInstance()->GetMaxQueueSize() );

	// Cache 생성 요청 정보 관리 queue
	CDataQueue<CCacheRequestData> queueCacheRequest;
	queueCacheRequest.SetMaxSize( CServiceConfig::GetInstance()->GetMaxQueueSize() );

	// Cache 생성 완료 결과 관리 queue
	CDataQueue<CCacheData> queueCacheComplete;
	queueCacheComplete.SetMaxSize( CServiceConfig::GetInstance()->GetMaxQueueSize() );

	// Cache 생성 관련 job 정보 관리 객체 생성
	CCacheJobList objCacheJob;


	// 1.2 공통 모듈 생성
	
	// 모니터링 및 전체 Thread 갯수 제어 관련 객체 생성	 
	CProcessStatus objProcessStatus( &queueReplicate, &queueRetry, &queueComplete, &queueCacheRequest, &queueCacheComplete, &objCacheJob );
	if( objProcessStatus.Init( CServiceConfig::GetInstance()->GetMaxReplicationThread()
		, CServiceConfig::GetInstance()->GetMaxCacheThread() ) == false )
	{
		LOG( LERR, "Worker[%d] process status module Init() failed.", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지
	}

	// RCDB 에서 proper FHS 정보를 조회하는 Thread 모듈 생성.
	CProperFhs objProperFhs;
	if( objProperFhs.Init() == false )
	{
		LOG( LERR, "Worker[%d] proper fhs management module Init() failed.", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지
	}
	

	// 1.3 Replication 관련 모듈 생성

	// replication 에 완료된 content 를 RCDB 에 저장 처리하는 thread 모듈 생성
	CReplicateInsert objReplicateInsert( &queueComplete );
	if( objReplicateInsert.Init() == false )
	{
		LOG( LERR, "Worker[%d] replication complete insert module Init() failed.", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지
	}
 	 
	// RCDB 상에서 replication 대상 content 조회 Thread 모듈 생성
	CReplicateSelect objReplicateSelect( &queueReplicate );	
	if( objReplicateSelect.Init() == false )
	{
		LOG( LERR, "Worker[%d] replication select module Init() failed.", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지		
	}

	// Replication Control Thread 모듈 생성
	CReplicateControl objReplicateControl( &objProperFhs, &objProcessStatus
		, &queueReplicate, &queueRetry, &queueComplete );
	if( objReplicateControl.Init() == false )
	{
		LOG( LERR, "Worker[%d] replication control module Init() failed.", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지
	}
		
	
	// 1.4 Hot Content Cache 관련 모듈 생성
	
	// RCDB 에서 유효기간이 만료된 Cache Content 삭제 Thread 모듈 생성.
	CCacheDelete objCacheDelete;
	if( objCacheDelete.Init() == false )
	{
		LOG( LERR, "Worker[%d] cache delete module Init() failed.", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지
	}

	// 생성된 Cache 완료 정보를 RCDB 에 저장하는 Thread 모듈 생성.
	CCacheInsert objCacheInsert( &queueCacheComplete, &objCacheJob );
	if( objCacheInsert.Init() == false )
	{
		LOG( LERR, "Worker[%d] cache complete insert module Init() failed.", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지
	}

	// Cache 요청을 확인하여 Job 에 등록을 수행하는 Thread 모듈 생성
	CCacheOrder objCacheOrder( &queueCacheRequest, &objCacheJob, &objProperFhs );
	if( objCacheOrder.Init() == false )
	{
		LOG( LERR, "Worker[%d] cache job order module Init() failed.", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지
	}
	
	// Cache 생성 처리를 담당하는 Thread 제어 모듈 생성
	CCacheControl objCacheControl( &objProcessStatus, &objCacheJob, &queueCacheComplete );
		

	// 2. 각 수행 모듈 기동 처리.
	//   - 기동 실패시 exit( EXIT_FAILURE ) 또는 return EXIT_FAILURE 를 수행하여
	//     초기화 실패로 인한 프로세스 종료가 발생토록 한다.

	// 2.1 공통 모듈 기동

	// Process 상태 표시 관련 Thread 기동
	if( objProcessStatus.Start() == false )
	{
		LOG( LERR, "Worker[%d]: Process status module start fail. => Worker Exit", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지		
	}
	_LOG( LINF, "Worker[%d]: CProcessStatus module start.", getpid() );
	
	// RCDB 에서 Proper FHS 내역 조회 및 관리 관련 모듈 기동처리.
	if( objProperFhs.Start() == false )
	{
		LOG( LERR, "Worker[%d]: Proper FHS management module start fail. => Worker Exit", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지		
	}
	_LOG( LINF, "Worker[%d]: CProperFhs module start.", getpid() );

	
	// 2.2 Replication 관련 수행 모듈 기동

	// Replication 완료 content 에 대한 RCDB insert 처리 모듈 기동처리.
	if( objReplicateInsert.Start() == false )
	{
		LOG( LERR, "Worker[%d]: Replicate complete insert module start fail. => Worker Exit", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지
	}
	_LOG( LINF, "Worker[%d]: CCompleteTargetInserter module start.", getpid() );
	
	// Replication 대상 content 조회 모듈 기동처리.
	if( objReplicateSelect.Start( CServiceConfig::GetInstance()->GetTargetSelectInterval()
		, CServiceConfig::GetInstance()->GetTargetSelectIntervalOnStarting() ) == false )
	{
		LOG( LERR, "Worker[%d]: Replicate target selector module start fail. => Worker Exit", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지		
	}
	_LOG( LINF, "Worker[%d]: CReplicateTargetSelector module start.", getpid() );
			
	// Replication 생성 관련 제어 모듈 기동처리.
	if( objReplicateControl.Start() == false )
	{
		LOG( LERR, "Worker[%d]: Replication control module start fail. => Worker Exit", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지		
	}
	_LOG( LINF, "Worker[%d]: CReplicateControl module start.", getpid() );

	
	// 2.3 Hot content cache 관련 수행 모듈 기동
	
	// RCDB 에서 유효기간 만료된 Cache Content 삭제 처리 모듈 기동처리.
	if( objCacheDelete.Start() == false )
	{
		LOG( LERR, "Worker[%d]: Cache delete module start fail. => Worker Exit", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지		
	}
	_LOG( LINF, "Worker[%d]: CCacheDelete module start.", getpid() );

	// 생성된 Cache 정보를 RCDB 에 저장하기 위한 모듈 기동 처리
	if( objCacheInsert.Start() == false )
	{
		LOG( LERR, "Worker[%d]: Complete cache insert module start fail. => Worker Exit", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지		
	}
	_LOG( LINF, "Worker[%d]: CCacheInsert module start.", getpid() );
		
	// Cache 요청을 확인하여 Job 에 등록을 수행하는 Thread 모듈 기동 처리
	if( objCacheOrder.Start() == false )
	{
		LOG( LERR, "Worker[%d]: Cache job order module start fail. => Worker Exit", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지		
	}
	_LOG( LINF, "Worker[%d]: CCacheOrder module start.", getpid() );

	// Cache 생성 Thread 제어 모듈 기동 처리
	if( objCacheControl.Start() == false )
	{
		LOG( LERR, "Worker[%d]: Cache control module start fail. => Worker Exit", getpid() );
		return EXIT_FAILURE; // EXIT_FAILURE 를 반환하여 재생성 처리 방지		
	}
	_LOG( LINF, "Worker[%d]: CCacheControl module start.", getpid() );


	// 모듈 초기화 및 기동 처리 완료.
	/////////////////////////////////////////////////////////////////////
	// worker 프로세스의 작업 수행.

	
	// fimngd 로 부터 hot content 정보를 수신 처리한다.
	int clientSocket;
	struct sockaddr_in clientSockAddr;
	socklen_t clientSockLen = sizeof( clientSockAddr );

	pthread_t clientThreadId;
	struct stFhsReceiveThreadArg * pThreadArg = NULL;

	while( 1 )
	{
		clientSocket = accept( socketListenHotContent, ( struct sockaddr * ) &clientSockAddr, &clientSockLen );
		if( clientSocket == -1 )
		{
			// 오류발생시 해당 내역 로깅처리.
			int errorNum = errno;
			LOG( LERR, "Worker: fimngd client accept failed. [%d][%s]", errorNum, strerror( errorNum ) );
		}
		else
		{
			// 정상적인 Client 인 경우...

			// Thread 로 전달할 변수를 malloc 로 생성 처리
			pThreadArg = ( struct stFhsReceiveThreadArg * ) malloc( sizeof( struct stFhsReceiveThreadArg ) );
			if( pThreadArg == NULL )
			{
				int errorNum = errno;
				close( clientSocket );
				
				char tempBuffer[INET_ADDRSTRLEN];
				memset( tempBuffer, 0x00, sizeof( tempBuffer ) );
				if( inet_ntop( AF_INET, &( clientSockAddr.sin_addr ), tempBuffer, sizeof( tempBuffer ) ) != NULL )
				{
					LOG( LERR, "[%s] hot content receive thread malloc failed [%d][%s]", tempBuffer, errorNum, strerror( errorNum ) );
				}
				else
				{
					LOG( LERR, "hot content receive thread malloc failed [%d][%s] and inet_ntop error[%d][%s]"
						, errorNum, strerror( errorNum ), errno, strerror( errno ) );
				}

				continue;
			}
			else
			{
				// 인자로 전달할 변수정보 설정.
				pThreadArg->clientSocketDesc   = clientSocket;
				pThreadArg->pQueueCacheRequest = &queueCacheRequest;

				char tempBuffer[INET_ADDRSTRLEN];
				memset( tempBuffer, 0x00, sizeof( tempBuffer ) );
				if( inet_ntop( AF_INET, &( clientSockAddr.sin_addr ), tempBuffer, sizeof( tempBuffer ) ) != NULL )
				{
					memcpy( pThreadArg->szClientAddress, tempBuffer, INET_ADDRSTRLEN );
				}
				else
				{
					snprintf( pThreadArg->szClientAddress, INET_ADDRSTRLEN - 1, "0.0.0.0" );
				}
				
				// fimngd 로부터 hot content 정보를 수신할 Thread 생성 처리
				if( pthread_create( &clientThreadId, NULL, threadHotContentReceive, (void *)pThreadArg ) != 0 )
				{
					// 생성 실패시...
					// - 로깅 
					// - client socket close 
					// - malloc 로 생성된 전달 변수 free 처리.

					int errorNum = errno;
					LOG( LERR, "[%s] hot content receive thread create failed [%d][%s]"
						, pThreadArg->szClientAddress, errorNum, strerror( errorNum ) );

					close( clientSocket );
					free( pThreadArg );
				}
			}
		}
	} // while( 1 )
	
    
	// Process 종료 처리.
	_LOG(  LINF, "Worker[%d] exit.. Good Bye..", getpid());
        
    //잠시 대기 후 종료처리.
	
	// CHG 2015-09-14 huibong
	// - #24369 usleep -> nanosleep 으로 변경 처리.
	struct timespec sleep;
	sleep.tv_sec = 0;
	sleep.tv_nsec = 500000000;  // 0.5 sec
	nanosleep( &sleep, NULL );


	// fork 를 수행한 함수에서 exit 함수 호출을 통한 종료처리
	return EXIT_SUCCESS;	
}


// FHS 의 fimngd 로 부터 TCP 연결 요청을 받아...
// Hot Content 정보를 수신 처리하기 위한 Thread 
void * threadHotContentReceive( void * arg )
{
	// 전달받은 변수를 내부 변수에 저장 처리.

	// client socket 관련 처리 변수 생성	
	CFimngdSocketControl clientSocket( ( ( struct stFhsReceiveThreadArg * ) arg )->clientSocketDesc );
	
	// client 주소 처리 관련 
	char szClientAddress[INET_ADDRSTRLEN];
	memcpy( szClientAddress, ( ( struct stFhsReceiveThreadArg * ) arg )->szClientAddress, INET_ADDRSTRLEN );

	// 수신된 정보를 저장하기 위한 queue 에 대한 포인터
	CDataQueue<CCacheRequestData> * pQueueCacheRequest = ( ( struct stFhsReceiveThreadArg * ) arg )->pQueueCacheRequest;
		
	// malloc 로 생성된 변수 해제 처리
	free( arg );

	// 본 thead 는 client 연결 상태에 따라 종료되므로.. process 에서 분리 처리
	pthread_detach( pthread_self() );

	_LOG( LDBG, "hot content recieve thread start [%s]", szClientAddress );
	

	// FHS fimngd 로 부터 data 수신 대기
	int nRead;
	CCacheRequestData receiveData;

	while( 1 )
	{
		// Data 수신 객체 초기화.
		receiveData.Init();

		// Data 수신
		nRead = clientSocket.GetData( receiveData );

		if( nRead == 0 )
		{
			// timeout 발생시			
			// - fimngd 는 data 전송 후.. 추가 전송할 data 가 있는지 일정기간 세션을 유지함.
			// - 따라서 .. 잠깐 sleep 후.. 다시 수신 시도.한다.
			sleep( 3 );
			continue;			
		}
		else if( nRead < 0 )
		{
			// Socket 오류 발생시
			// - 관련 로그는 해당 모듈에서 다 찍었으므로.. 여기서는 루프 종료 처리.
			break;
		}
		else
		{
			// 정상 Data 수신시 유효 요청인지 검사한다.
			if( CLogger::GetInstance() != NULL && CLogger::GetInstance()->GetLogLevel() >= LDBG )
			{
				receiveData.Print();
			}

			// 1. RCID 확인
			if( strcasecmp( receiveData.m_strRcid.c_str(), CServiceConfig::GetInstance()->GetRcId() ) != 0 )
			{
				// RCID 가 유효하지 않은 경우..
				// - 잘못된 세션 요청이므로...  
				// - 로깅, Report 로깅 후 세션 종료 처리한다.
				_LOG( LWAR, "cache request: rc id mismatch.[%s][%s]"
					, receiveData.m_strRcid.c_str()
					, receiveData.m_strRequestHost.c_str() );

				CReportLog reportLog;
				if( reportLog.Init( CServiceConfig::GetInstance()->GetLogPath() ) == false )
				{
					std::string strErrorMessage;
					reportLog.GetLastError( strErrorMessage );

					LOG( LERR, "report log init fail. [%s]", strErrorMessage.c_str() );
				}
				else
				{
					reportLog.Write( "ERR", "[%s][Cache request RC ID mismatch. check FHS[%s][%s]]"
						, PROG_NAME
						, receiveData.m_strRequestHost.c_str()
						, receiveData.m_strRcid.c_str() );
				}

				// 세션 종료 처리.
				break;
			}
			
			// 2. filename_hash 정보가 정확한지
			if( receiveData.m_strFileNameHash.empty() == true )
			{
				// filename_hash 정보가 존재하지 않는 경우..
				// - 현재 수신된 정보는 버리고.. 다시 수신 대기
				_LOG( LWAR, "cache request: file name empty. [%s]"
					, receiveData.m_strRequestHost.c_str() );

				// 다시 수신 대기
				continue;
			}
			// 3. szServiceID ( 구 Tran ID 정보) 가 정확한지 검사			
			if( receiveData.m_strServiceSeq.empty() == true )
			{
				// 서비스 ID  정보가 존재하지 않는 경우..
				// - 현재 수신된 정보는 버리고.. 다시 수신 대기
				_LOG( LWAR, "cache request: service sequence empty. [%s]"
					, receiveData.m_strRequestHost.c_str() );

				// 다시 수신 대기
				continue;
			}

			
			// 4. Content Length 값이 0 이상인지....
			if( receiveData.m_uContentLength == 0 )
			{
				_LOG( LWAR, "cache request: content length 0. [%s][%s]"
					, receiveData.m_strRequestHost.c_str()
					, receiveData.m_strFileNameHash.c_str() );
				
				// 다시 수신 대기
				continue;
			}

			// 5. 동접 FD 의 값이 conf 에 설정된 값 이상인지 검사
			if( receiveData.m_uConcurrentFd < CServiceConfig::GetInstance()->GetHotContentCreateMinimumFd() )
			{
				// Cache 생성 요청의 동접 FD 값이 conf 의 설정값 미만인 경우....
				// - 다음 data 수신 대기
				_LOG( LDBG, "cache request: fd count lower. ignored. [%u/%u] [%s][%s]"
					, receiveData.m_uConcurrentFd
					, CServiceConfig::GetInstance()->GetHotContentCreateMinimumFd()
					, receiveData.m_strRequestHost.c_str()
					, receiveData.m_strFileNameHash.c_str() );

				continue;
			}
			
			// 유효 요청인 경우... 
			// - Queue 에 Push 처리한다.
			// - 만약 queue 가 full 이라 push 가 실패나는 경우.. 3번 재시도, 안되면 버린다.
			int nRetry = 0;
			while( pQueueCacheRequest->Push( receiveData ) == false )
			{
				++nRetry;

				if( nRetry > 3 )
				{
					_LOG( LWAR, "Cache request push to queue fail. ignored [%s][%s]"
						, receiveData.m_strRequestHost.c_str()
						, receiveData.m_strFileNameHash.c_str() );

					break;
				}
				else
				{
					sleep( 1 );
				}
			}
		}
	}

	// client 와 연결 종료 처리.
	_LOG( LDBG, "hot content recieve connection closed.[%s]", szClientAddress );

	clientSocket.Close();

	return NULL;
}



/// @brief Worker Process 종료 Signal 을 전달받은 경우 이를 처리하기 위한 함수.
/// @param nSignalNumber [in] 발생한 시그널 Number
/// @return void
static void SignalWorkerTerminate(int nSignalNumber)
{
	// Signal Number 에 따른 로깅처리.
	if( nSignalNumber == SIGTERM )
	{
		_LOG( LINF, "Worker[%d] process exit by user signal [SIGTERM]. Good Bye..", getpid() );
	}
	else
	{
		_LOG( LWAR, "Worker[%d] process exit by abnormal signal[%d], Good Bye..", getpid(), nSignalNumber );
	}

	// CHG 2015-09-14 huibong
	// - #24369 usleep -> nanosleep 으로 변경 처리.
	struct timespec sleep;
	sleep.tv_sec = 0;
	sleep.tv_nsec = 500000000;  // 0.5 sec
	nanosleep( &sleep, NULL );
	

	exit( EXIT_SUCCESS );
}

/// @brief Worker 프로세스 signal 처리 설정을 위한 함수	
/// @return void	
void SetSignalWorker( void )
{
	sigset_t set;
	struct sigaction act;
	memset(&act, 0x00, sizeof(act));

	sigfillset( &set );
	sigprocmask( SIG_SETMASK, &set, NULL );	
	sigfillset( &act.sa_mask );

	/* 무시할 신호 목록 */
	act.sa_handler = SIG_IGN;
	sigaction( SIGPIPE, &act, NULL);	/* desciptor 오류 발생시 Process가 죽는 것은 방지하기 위하여 설정 */
	sigaction( SIGHUP,  &act, NULL);	/* process를 기동시킨 관리자의 로그아웃시, 또는 config reload 등의 처리 signal*/
	sigaction( SIGINT,  &act, NULL);	/* ^C 키를 누른 경우 받는 신호 => demon 으로 기동되기 땜시 이 신호 못받음 */
	sigaction( SIGQUIT, &act, NULL);	/* 키보드에 의한  Abort 신호 처리 => ? */
	
	// Worker 프로세스는 자식 프로세스가 존재하지 않으므로 그냥 무시처리함.
	act.sa_handler = SIG_IGN;
	sigaction( SIGCHLD, &act, NULL);
	
	/* 각종 에러나 사용자의 종료 신호 처리 */
	act.sa_handler = SignalWorkerTerminate;
	sigaction( SIGTERM, &act, NULL);	/* kill -TERM 에 의한 프로세스 종료시 */

	sigemptyset(&set);                  /* 신호 처리기 처리 설정 위한 블록 해제 */
	sigprocmask(SIG_SETMASK, &set, NULL);
}
