398 lines
12 KiB
C++
398 lines
12 KiB
C++
#include "ReplicateThread.h"
|
|
|
|
#include <errno.h>
|
|
#include <time.h>
|
|
#include <string.h>
|
|
|
|
#include "Logger.h"
|
|
#include "FtsdSocketControl.h"
|
|
#include "ServiceConfig.h"
|
|
#include "ReportLog.h"
|
|
|
|
|
|
// ftsd 로 부터 처리 결과 수신을 위한 Total 대기 시간 정의 ( sec )
|
|
#define FHS_RESPONSE_TOTAL_TIMEOUT 3600 // ( sec, 1 hour )
|
|
|
|
// ftsd 로 부터 응답 수신 확인을 위한 socket read timeout (sec)
|
|
#define FHS_RESPONSE_CHECK_TIMEOUT 3 // 3 sec
|
|
|
|
////////////////////////////////////////////
|
|
|
|
// 생성자
|
|
CReplicateThread::CReplicateThread( CProcessStatus* const pProcessStatus
|
|
, CDataQueue<CRetryData>* const pQueueRetry
|
|
, CDataQueue<CCompleteData>* const pQueueComplete
|
|
, CRetryData& dataJob
|
|
, std::vector<std::string>& vecTargetFHS )
|
|
: m_pProcessStatus( pProcessStatus )
|
|
, m_pRetryQueue( pQueueRetry )
|
|
, m_pCompleteQueue( pQueueComplete )
|
|
, m_dataJob( dataJob )
|
|
, m_vecTargetFHS( vecTargetFHS )
|
|
{
|
|
// 멤버 변수 초기화
|
|
m_threadHandle = 0;
|
|
}
|
|
|
|
// 소멸자
|
|
CReplicateThread::~CReplicateThread()
|
|
{
|
|
// Thread 동작 정지 처리
|
|
// - 만약 Thread 가 이미 종료된 경우 m_threadHandle 이 다른 Thread Handle 일 수 있으므로
|
|
// 업무 Flow 수정시 주의할 것
|
|
if( m_threadHandle != 0 )
|
|
pthread_cancel( m_threadHandle );
|
|
}
|
|
|
|
// 실제 Replication 생성 작업을 수행
|
|
bool CReplicateThread::Start()
|
|
{
|
|
int nRet = ::pthread_create( &m_threadHandle, NULL, CReplicateThread::threadFunc, this );
|
|
if( nRet != 0 )
|
|
{
|
|
// Thread 생성 실패시
|
|
int errorNum = errno;
|
|
LOG( LERR, "Thread create failed.[%d][%s]", errorNum, strerror( errorNum ) );
|
|
|
|
return false;
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
// 스레드 함수
|
|
void* CReplicateThread::threadFunc( void* arg )
|
|
{
|
|
CReplicateThread* pObject = reinterpret_cast<CReplicateThread *>( arg );
|
|
pthread_detach( pthread_self() );
|
|
|
|
pthread_testcancel();
|
|
// Thread 동작시 Thread Count 1 증가 처리
|
|
pObject->m_pProcessStatus->AddReplicationCount();
|
|
|
|
// 실제 작업 수행.
|
|
pObject->Execute();
|
|
|
|
// Thread 동작 완료시 Thread Count 1 감소 처리
|
|
pObject->m_pProcessStatus->SubtractReplicationCount();
|
|
pthread_testcancel();
|
|
|
|
|
|
// NEW 2014-06-25 huibong
|
|
// Thread 종료시 m_threadHandle 값을 초기화 처리.
|
|
// - 소멸자에서 잘못된 Thread 를 종료할 수 있기 때문.
|
|
pObject->m_threadHandle = 0;
|
|
|
|
// 객체 자동 delete 처리.
|
|
delete pObject;
|
|
return NULL;
|
|
}
|
|
|
|
// 실제 Replication 생성 작업을 수행
|
|
void CReplicateThread::Execute()
|
|
{
|
|
// 1. FHS 의 ftsd 와 통신 처리를 담당할 socket control 객체 생성.
|
|
CFtsdSocketControl fhs;
|
|
|
|
// 2. 원본 Content를 가진 FHS ftsd 에 접속 수행.
|
|
bool bResult = fhs.ConnectTarget( m_dataJob.m_OriginData.m_szOriginHostName, CServiceConfig::GetInstance()->GetFhsTransferDaemonPort());
|
|
if( bResult == false )
|
|
{
|
|
// 원본 FHS 와 접속 실패시
|
|
// - 로깅 후 Retry Queue 에 저장 처리
|
|
_LOG( LERR, "CReplicateThread: source fhs[%s] connect fail.[%s]"
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.c_str() );
|
|
|
|
DoFailProcess();
|
|
return;
|
|
}
|
|
|
|
// 3. 연결된 원본 FHS 로 replication 생성 요청을 전달
|
|
bResult = fhs.SendFileReplicationRequest( m_dataJob.m_OriginData.m_szFileNameHash
|
|
, m_dataJob.m_OriginData.m_nContentLength
|
|
, CServiceConfig::GetInstance()->GetHashCheckLevel()
|
|
, true
|
|
, m_vecTargetFHS );
|
|
if( bResult == false )
|
|
{
|
|
// replication 요청 전달 실패시
|
|
// - 로깅 후 Retry Queue 에 저장 처리
|
|
_LOG( LERR, "CReplicateThread: replication request send fail.[%s][%s]"
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.c_str() );
|
|
|
|
DoFailProcess();
|
|
return;
|
|
}
|
|
|
|
// 4. FHS 로 부터 처리 결과를 수신할때 까지 대기
|
|
// - 이상 상황 발생시 무한 대기 가능성이 존재하므로..
|
|
// 미리 정의된 FHS_RESPONSE_TOTAL_TIMEOUT 까지만 대기하도록 한다.
|
|
|
|
// response timeout 처리를 위한 변수
|
|
time_t timeStart;
|
|
time_t timeNow;
|
|
|
|
// 응답 결과를 저장하기 위한 변수
|
|
int nResult;
|
|
bool bSuccess = false;
|
|
std::string szErrorMessage;
|
|
|
|
// 대기 시작 전 현재시간을 구한다.
|
|
timeStart = time( &timeStart );
|
|
|
|
// FHS 로 부터 응답 수신 대기
|
|
while( 1 )
|
|
{
|
|
// FHS 로 부터 응답 수신 처리
|
|
// -1 : 연결 종료
|
|
// 0 : Timeout 발생
|
|
// 1 : 응답 수신
|
|
// 2 : 복제 관련 응답이 아닌 다른 응답인 경우.. (발생하지 않지만.. 혹시나)
|
|
nResult = fhs.GetFileReplicationResult( FHS_RESPONSE_CHECK_TIMEOUT
|
|
, bSuccess
|
|
, szErrorMessage
|
|
, m_mapSuccessFhs
|
|
, m_mapFailFhs );
|
|
|
|
if( nResult == 0 )
|
|
{
|
|
// 지정된 시간동안 응답 결과를 수신하지 못한 경우
|
|
// - FHS 와 연결이 정상인지 Alive 체크 호출
|
|
// - 만약 연결이 끝어진 경우.. 다음 loop 에서 -1 연결 종료가 리턴됨.
|
|
if( fhs.SendAliveCheck() == false )
|
|
{
|
|
_LOG( LERR, "CReplicateThread: alive check fail.[%s][%s]"
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.c_str() );
|
|
}
|
|
|
|
// Timeout 이 발생했으므로 바로 재시도하지 않고 조금 대기 후 다시 시도.
|
|
sleep( 1 );
|
|
}
|
|
else if( nResult == 1 )
|
|
{
|
|
// FHS 로 부터 응답을 수신한 경우.
|
|
// - Content 이상으로 인해 전송 자체를 하지 못했는지 여부 판단
|
|
if( bSuccess == true )
|
|
{
|
|
// 정상적으로 Replication 이 처리된 경우..
|
|
_LOG( LDBG, "CReplicateThread: result receive.[%s][%s]"
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.c_str() );
|
|
|
|
DoSuccessProcess();
|
|
}
|
|
else
|
|
{
|
|
// Content 이상으로 인해 Replication 자체에 문제가 있는 경우.
|
|
_LOG( LERR, "CReplicateThread: error result receive.[%s][%s][%s][%llu]"
|
|
, szErrorMessage.c_str()
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.c_str()
|
|
, m_dataJob.m_OriginData.m_nContentLength );
|
|
|
|
DoErrorProcess( szErrorMessage );
|
|
}
|
|
|
|
// 처리 결과를 수신했으므로 loop 종료.
|
|
break;
|
|
}
|
|
else if( nResult == -1 )
|
|
{
|
|
// FHS 와 연결이 끝어진 경우.
|
|
// - retry queue 에 저장하고 본 루프는 종료
|
|
_LOG( LERR, "CReplicateThread: fhs connection broken.[%s][%s]"
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.c_str() );
|
|
|
|
DoFailProcess();
|
|
break;
|
|
}
|
|
else if( nResult == 2 )
|
|
{
|
|
// Replication 관련 응답이 아닌 경우
|
|
// - 무시
|
|
;
|
|
}
|
|
else
|
|
{
|
|
// 기타 정의되지 않은 결과값 반환시
|
|
// - FHS 와 세션 이상이 발생할 수 있으므로
|
|
// - 실패로 처리하여 retry queue 에 저장하고 본 루프는 종료
|
|
_LOG( LERR, "CReplicateThread: receive undefine result.[%d][%s][%s]"
|
|
, nResult
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.c_str() );
|
|
|
|
DoFailProcess();
|
|
break;
|
|
}
|
|
|
|
// Total Reponse 대기 시간 확인
|
|
if( difftime( time( &timeNow ), timeStart ) > FHS_RESPONSE_TOTAL_TIMEOUT )
|
|
{
|
|
// Total Response Timeout 발생시
|
|
// - 로깅 후 Retry Queue 에 저장 처리
|
|
// - 본 대기 루프를 종료
|
|
_LOG( LERR, "CReplicateThread: replication result receive timeout.[%s][%s]"
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.c_str() );
|
|
|
|
DoFailProcess();
|
|
break;
|
|
}
|
|
|
|
} // while( 1 )
|
|
|
|
}
|
|
|
|
// Replication 처리 관련 오류가 발생한 경우 처리 함수.
|
|
// - 재시도 횟수를 확인하여
|
|
// - Retry queue 에 저장 처리
|
|
void CReplicateThread::DoFailProcess()
|
|
{
|
|
// 재시도 횟수 초과 여부 확인.
|
|
if( m_dataJob.m_uFailCount + 1 > CServiceConfig::GetInstance()->GetMaxRetryCount() )
|
|
{
|
|
// 재시도 횟수 초과시
|
|
// - Report 로그에 기록하고 Retry Queue 에는 저장하지 않는다.
|
|
_LOG( LERR, "CReplicateThread: replicate thread fail and retry count over.[%s][%s]"
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.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][Replication fail. check content[%s][%s]]"
|
|
, PROG_NAME
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.c_str() );
|
|
}
|
|
}
|
|
else
|
|
{
|
|
// Retry Queue 에 저장 처리.
|
|
m_dataJob.m_uFailCount += 1;
|
|
m_dataJob.m_timeFailed = time( 0 );
|
|
|
|
// Retry Queue 가 Full 인 상태에서
|
|
// 작업 Thread 가 모두 Push 상태인 경우 무한 Block 상태 발생 가능...
|
|
// 하지만.. 실제 발생 가능이 희박하므로... 무한 재시도 처리.
|
|
while( m_pRetryQueue->Push( m_dataJob ) == false )
|
|
{
|
|
sleep( 1 );
|
|
}
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
|
|
// Replication 처리에 대해 FHS 에서 실패로 응답한 경우.
|
|
// - 대부분 Replication 대상 content 가 존재하지 않거나 Size 변경 등이 발생한 경우임.
|
|
// - 따라서 해당 Content 에 대해서는 Retry 하지 않고.. Report 로그에만 기록하고 끝낸다.
|
|
void CReplicateThread::DoErrorProcess( string& szErrorMessage )
|
|
{
|
|
// 2017-11-15 CHG huibong 대량 report 로그 발생 관련 항목 조정 (#31526)
|
|
// - 토토디스크 서비스에서 업로드에 따른 size mismatch 대량 report 로그 관련
|
|
// - error message 가 "source file size not match" 인 경우 report 로그에 기록되지 않도록 수정한다.
|
|
// - 실제 error message 내용은 "source file size not match.[ request: 877314616, local: 1137364384 ]" 형식이므로
|
|
// - 앞의 일부분만 비교 처리.
|
|
if(strncmp(szErrorMessage.c_str(), "source file size not match", 26) == 0 )
|
|
{
|
|
return;
|
|
}
|
|
else
|
|
{
|
|
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][Replication fail. check content[%s][%s][%llu][%s]]"
|
|
, PROG_NAME
|
|
, m_dataJob.m_OriginData.m_szOriginHostName.c_str()
|
|
, m_dataJob.m_OriginData.m_szURI.c_str()
|
|
, m_dataJob.m_OriginData.m_nContentLength
|
|
, szErrorMessage.c_str() );
|
|
}
|
|
|
|
return;
|
|
}
|
|
}
|
|
|
|
|
|
// Replication 처리에 대해 FHS 에서 성공을 응답한 경우 처리함수
|
|
// - 성공된 FHS 와 실패된 FHS 정보를 동시에 전달하므로 확인하여 처리해야 한다.
|
|
void CReplicateThread::DoSuccessProcess()
|
|
{
|
|
// 멤버 변수에 저장된 정보를 처리하기 위한 iterator
|
|
std::map<std::string, std::string>::iterator itSuccessFhs;
|
|
|
|
// m_mapSuccessFhs 에 저장된 성공 결과에 대한 처리
|
|
// - success fhs host, hash_name 의 정보 저장된 상태
|
|
// - Complete Queue 에 저장.
|
|
for( itSuccessFhs = m_mapSuccessFhs.begin(); itSuccessFhs != m_mapSuccessFhs.end(); ++itSuccessFhs )
|
|
{
|
|
// complete queue 에 저장하기 위한 객체 조립
|
|
CCompleteData completeResult;
|
|
completeResult.m_OriginData = m_dataJob.m_OriginData;
|
|
completeResult.m_szTargetHostName = itSuccessFhs->first;
|
|
completeResult.m_szTargetFileNameHash = itSuccessFhs->second;
|
|
|
|
// complete queue 에 저장 처리
|
|
// - complete queue 가 full 상태일 수 있어 push 실패 가능성 존재
|
|
// - 하지만... complete queue 에 저장 못할 경우...
|
|
// FHS 에 생성된 replication 은 처치 곤란한 쓰레기가 될 수 있으므로....
|
|
// 최대한 push 처리한다.
|
|
while( m_pCompleteQueue->Push( completeResult ) == false )
|
|
{
|
|
// push 실패시 잠시 대기 후 재시도
|
|
sleep( 1 );
|
|
}
|
|
|
|
// 원본 Data 객체에 성공한 FHS 정보 입력 처리
|
|
// - 일부 FHS 에만 복제 성공하고... 다른 FHS 복제는 실패한 경우...
|
|
// - Retry 를 통해 복제가 재시도 되도록 하기 위함.
|
|
m_dataJob.m_listSuccessHost.push_back( completeResult.m_szTargetHostName );
|
|
}
|
|
|
|
// m_mapFailFhs 에 저장된 실패 결과에 대한 처리
|
|
// - 재시도 횟수 초과시 Report 로그 기록 후 버림
|
|
// - 재시도 횟수 이하인 경우 fail count 증가 후 Retry Queue 에 저장
|
|
if( m_mapFailFhs.empty() == false )
|
|
{
|
|
// 멤버 변수에 저장된 정보를 처리하기 위한 iterator
|
|
// - failed fhs host, error message
|
|
std::map<std::string, std::string>::iterator itFailFhs;
|
|
|
|
// 실패 내역에 대한 로깅
|
|
for( itFailFhs = m_mapFailFhs.begin(); itFailFhs != m_mapFailFhs.end(); ++itFailFhs )
|
|
{
|
|
_LOG( LWAR, "CReplicateThread: replicate fail [%s]->[%s][%s]"
|
|
, m_dataJob.m_OriginData.m_szURI.c_str()
|
|
, itFailFhs->first.c_str()
|
|
, itFailFhs->second.c_str() );
|
|
}
|
|
|
|
// 재시도 횟수 확인 처리 및 Retry queue 관련 처리
|
|
DoFailProcess();
|
|
}
|
|
|
|
return;
|
|
}
|