#include "CacheOrder.h" #include #include #include "Database.h" #include "Logger.h" // 처리 loop 간의 sleep time 정의 #define DEFAULT_SLEEP_TIME 3 // RCDB 접속 실패 또는 Query 실패 등의 상황 발생시 #define MAX_SLEEP_TIME 30 // Query 문장 저장을 위한 Buffer 크기 #define DEFAULT_QUERY_BUFFER_SIZE 2048 // 생성자 // param [IN] pQueueCacheRequest Cache 요청 정보 저장 Queue 에 대한 포인터 // param [IN] pListCacheJob Cache 생성 관련 Job List 객체에 대한 포인터 // param [IN] pProperFhs Proper FHS 정보를 저장하는 객체에 대한 포인터 CCacheOrder::CCacheOrder( CDataQueue* pQueueCacheRequest, CCacheJobList* pCacheJob, CProperFhs * pProperFhs ) : m_pQueueCacheRequest( pQueueCacheRequest ) , m_pCacheJob( pCacheJob ) , m_pProperFhs( pProperFhs ) { m_threadHandle = 0; m_uSleep = DEFAULT_SLEEP_TIME; // NEW 2014-07-07 신규 멤버 변수 초기화 코드 추가. // Init() 에서 RCDB 접속 실패가 발생 가능하므로.. 구형 RCDB 형상을 사용토록 설정. m_bServiceTableSeparate = false; } // 소멸자 CCacheOrder::~CCacheOrder() { // Thread 동작 정지 처리 // - 만약 Thread 가 이미 종료된 경우 m_threadHandle 이 다른 Thread Handle 일 수 있으므로 // 업무 Flow 수정시 주의할 것 if( m_threadHandle != 0 ) pthread_cancel( m_threadHandle ); } // 객체 초기화 수행 bool CCacheOrder::Init() { // 생성자로 전달받은 각 객체에 대한 포인터는 유효하다가 판단하고.... // 각 멤버변수에 대한 검사는 수행하지 않는다. // RCDB 접속 정보 Load. m_rcdbInfo.Load(); // 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 ) { LOG( LERR, "RCDB connection failed." ); return false; } else { // 접속 성공한 경우... LOG( LINF, "RCDB connection test ok..." ); // NEW 2014-06-23 huibong // 서비스 table 분리된 신규 형상인지.. 구형 형상인지 체크하여 // m_bServiceTableSeparate 변수 설정 처리. // 해당 변수 초기화는 생성자에서 1차로 수행함. char szQuery[1024]; snprintf( szQuery, 1023, "SELECT * FROM pg_tables WHERE schemaname = 'public' AND tablename = 't_dav_resource' " ); rcdb.PgDoExec( szQuery ); if( rcdb.PgResult( DataBase::NOT_CLEAR ) < 0 ) { // Query 수행 결과 오류 발생시... // - pg_tables 가 존재하지 않을 경우 오류 발생 가능... // - 이는 RCDB 가 아직 적절하게 구성되지 않았다는 의미이므로... Init()함수를 오류로 처리한다. LOG( LERR, "RCDB t_dav_resource exist check failed.[%s][%s]", rcdb.GetErrorMessage().c_str(), szQuery ); rcdb.PgClear(); rcdb.PgCloseDB(); return false; } else { // Query 수행이 정상인 경우. int nResult = rcdb.GetNoTuples(); if( nResult > 0 ) { // t_dav_resource 테이블이 존재하는 경우... m_bServiceTableSeparate = false; _LOG( LINF, "CCacheOrder: t_dav_resource table exist.[Old RCDB Type]" ); } else { // t_dav_resource 테이블이 존재하지 않는 서비스별 테이블 분리 형상인 경우. m_bServiceTableSeparate = true; _LOG( LINF, "CCacheOrder: t_dav_resource table not exist.[New RCDB Type]" ); } } // RCDB 조회 결과 set 를 clear 처리 후 RCDB 접속 해제. rcdb.PgClear(); rcdb.PgCloseDB(); return true; } } // thread를 생성하여 업무 flow 를 시작. bool CCacheOrder::Start() { int nRet = ::pthread_create( &m_threadHandle, 0, CCacheOrder::threadFunc, this ); if( nRet != 0 ) { // Thread 생성 실패시 int errorNum = errno; LOG( LERR, "Thread create failed.[%d][%s]", errorNum, strerror( errorNum ) ); return false; } return true; } // 스레드 함수 void* CCacheOrder::threadFunc( void* arg ) { CCacheOrder* pObject = reinterpret_cast( arg ); pthread_detach( pthread_self() ); while( 1 ) { pthread_testcancel(); pObject->Execute(); pthread_testcancel(); // 1회 수행 후 지정된 시간 만큼 대기 sleep( pObject->m_uSleep ); } // NEW 2014-07-07 huibong // Thread 종료시 m_threadHandle 값을 초기화 처리. // - 소멸자에서 잘못된 Thread 를 종료할 수 있기 때문. pObject->m_threadHandle = 0; return NULL; } // 스레드 내에서 각 loop 단위 처리 호출 함수. void CCacheOrder::Execute() { // Request Cache Queue 상에 처리할 Data 가 존재하는 경우. if( m_pQueueCacheRequest->GetSize() > 0 ) { bool bResult = PushRequestToJob(); // 처리 결과에 따라 loop sleep time 조정 if( bResult == true ) m_uSleep = DEFAULT_SLEEP_TIME; else { // RCDB 연결 실패, Queury 수행 실패 등의 오류 발생시 m_uSleep = MAX_SLEEP_TIME; } } else { // 처리할 Data 가 존재하지 않는 경우. m_uSleep = DEFAULT_SLEEP_TIME; } } // Cache Requet Queue 에 저장된 요청을 검사하여.... // cache 생성 Job 에 추가하는 기능을 수행 bool CCacheOrder::PushRequestToJob() { // RCDB 연결 관련 객체. // - RCDB 연결은 Job 에 등록가능한 Data 가 존재할 경우에만 수행되도록 한다. DataBase rcdb; bool bDbConnected = false; // RCDB 연결 상태 정보 저장 변수 // Queue 에서 Data 를 Pop 한 후 // 해당 정보가 CacheJobList 에 이미 등록된 정보인지 확인 // 미등록 정보인 경우 RCDB 조회를 수행하여 // - Cache 를 수행할 FHS 정보를 선택한 후 // - 해당 Content 정보를 Cache Job 에 등록 처리한다. // queue 에서 꺼낸 정보를 저장하기 위한 변수 CCacheRequestData requestData; // Query 문장 저장을 위한 버퍼 생성 char szQuery[DEFAULT_QUERY_BUFFER_SIZE]; // Query 수행 결과값 저장을 위한 변수.. int nResult = 0; while( m_pQueueCacheRequest->GetSize() > 0 ) { // queue 에 저장된 첫번째 data 복사 (queue 에서 삭제된 상태 아님) requestData = m_pQueueCacheRequest->Front(); // 해당 정보를 Job List 에 저장 가능한지 확인 if( m_pCacheJob->IsExist( requestData ) == true ) { // 이미 동일 정보가 Job List 에 존재하는 경우... // queue 에서 pop 해서 버림. _LOG( LINF, "Cache request already exist.[%s]", requestData.m_strFileNameHash.c_str() ); m_pQueueCacheRequest->Pop(); continue; } // RCDB 에서 해당 content 에 대한 정보 조회 // DB 연결이 되지 않은 상태인 경우. // - Queue 항목 중 Job 에 등록해야 하는 항목이 존재할 경우에만 RCDB 연결 목적 if( bDbConnected == false ) { // 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 ) { // RCDB 접속이 안될때... // - 현재 Queue 에 저장된 Cache Request 요청을 모두 버리고 // - false 반환하여 특정시간 동안 sleep 되도록 처리. LOG( LERR, "RCDB connection failed. Cache request all trhow [%u]" , m_pQueueCacheRequest->GetSize() ); // Queue 요청 모두 삭제 처리 m_pQueueCacheRequest->Init(); return false; } else { bDbConnected = true; } } // Query 조립 // - 전달받은 filename_hash , content_length, FHS 정보를 이용하여 // - 해당 물리 파일과 연결된 uri 추출 ( uri 를 직접 수신할 경우..대소문자 무시 기능 때문에 조회 불가) // - 추출된 uri 와 연결된 장비 정보를 조회한다. // - ORDER BY 조건에 의해 is_cache 값이 N-R-Y 순으로 나타남. // - COPY 등으로 인해 Query 결과에는 중복된 host_name 이 존재 가능함. // t_dav_resource 테이블 사용 RCDB 인 경우 if( m_bServiceTableSeparate == false ) { sprintf( szQuery, "SELECT host_name, filename_hash, is_cache " "FROM t_dav_resource " "WHERE uri IN " " ( SELECT uri FROM t_dav_resource " " WHERE filename_hash = '%s' " " AND host_name = '%s' " " AND get_content_length = %llu " " AND deleted_yn = 'N' " " ) " " AND get_content_length = %llu " " AND deleted_yn = 'N' " "ORDER BY is_cache ASC; " , requestData.m_strFileNameHash.c_str() , requestData.m_strRequestHost.c_str() , requestData.m_uContentLength , requestData.m_uContentLength ); } else { // 신형 RCDB 형상인 경우. sprintf( szQuery, "SELECT host_name, filename_hash, is_cache " "FROM t_meta_%s " "WHERE uri IN " " ( SELECT uri FROM t_meta_%s " " WHERE filename_hash = '%s' " " AND host_name = '%s' " " AND get_content_length = %llu " " AND deleted_yn = 'N' " " ) " " AND get_content_length = %llu " " AND deleted_yn = 'N' " "ORDER BY is_cache ASC; " , requestData.m_strServiceSeq.c_str() , requestData.m_strServiceSeq.c_str() , requestData.m_strFileNameHash.c_str() , requestData.m_strRequestHost.c_str() , requestData.m_uContentLength , requestData.m_uContentLength ); } // Query 실행 및 결과 확인 rcdb.PgDoExec( szQuery ); nResult = rcdb.PgResult( DataBase::NOT_CLEAR ); if( nResult < 0 ) { // RCDB t_dav_resource 테이블 조회 실패시 // - RCDB 연결 실패 또는 Query 실패로 인한 문제 발생 가능. // - 따라서 RCDB 연결 해제 처리, Queue Pop 처리한 후 // - 루프를 계속 수행처리 LOG( LERR, "RCDB t_dav_resource cache select failed.[%s][%s]" , rcdb.GetErrorMessage().c_str(), szQuery ); rcdb.PgClear(); rcdb.PgCloseDB(); bDbConnected = false; m_pQueueCacheRequest->Pop(); continue; } // 결과 Row 수 확인. nResult = rcdb.GetNoTuples(); if( nResult <= 0 ) { // 해당 Content 가 존재하지 않거나... 오류 발생시. // - 그냥 버린다. // - 이로 인해 FHS 에 meta 기준으로 삭제 불가능한 파일이 남게 된다. _LOG( LINF, "cache request content not found in RCDB.[%d] [%s][%s][%llu]" , nResult , requestData.m_strRequestHost.c_str() , requestData.m_strFileNameHash.c_str() , requestData.m_uContentLength ); } else { // 결과 Row 수가 존재하는 경우... CCacheData cacheData; cacheData.m_requestData = requestData; // 결과 정보 중.. Request 요청을 보낸 장비가 아닌 FHS 을 Cache 생성 원본으로 선택 // - order by 조건에 의해 is_cache 가 N-R-Y 순으로 장비 정보가 나타나므로... // - 원본, 복제본, cache 본 순으로 사용될 순서가 정해짐. for( int i = 0; i < nResult; ++i ) { // host name 이 같지 않은 경우. if( strcasecmp( rcdb.GetValue( i, 0 ), requestData.m_strRequestHost.c_str() ) != 0 ) { cacheData.m_strSourceHost = rcdb.GetValue( i, 0 ); cacheData.m_strSourceFilenameHash = rcdb.GetValue( i, 1 ); // source 로 사용할 첫번째 장비를 찾은 경우... // - 더 이상 루프는 의미가 없으므로.. break; } } // 만약 원본 1개만 존재하는 경우... // - 그냥 요청 정보를 그대로 사용. if( cacheData.m_strSourceHost.empty() == true ) { cacheData.m_strSourceHost = requestData.m_strRequestHost; cacheData.m_strSourceFilenameHash = requestData.m_strFileNameHash; } // ignore FHS list 구성 // - 빠른 처리를 위해 조회 결과 처리시 중복 host_name 정보를 제거한다. std::list ignoreFhsList; std::string strTemp; for( int j = 0; j < nResult; ++j ) { strTemp = rcdb.GetValue( j, 0 ); ignoreFhsList.push_back( strTemp ); } // 중복 제거 ( sort() 후 unique() 함수 호출해야 중복 제거 가능 ) ignoreFhsList.sort(); ignoreFhsList.unique(); // Target FHS 선정 처리 std::vector vecProperFhs; if( m_pProperFhs->GetProperFhs( 1, vecProperFhs, ignoreFhsList ) == false ) { // Proper FHS 선택 실패시... // 해당 요청을 그냥 버린다... _LOG( LWAR, "cache request not found proper fhs. throw.[%s][%s][%llu]" , requestData.m_strRequestHost.c_str() , requestData.m_strFileNameHash.c_str() , requestData.m_uContentLength ); } else { // Proper FHS 선택 성공시.. // - 해당 정보를 CacheData 에 입력한 후... // - Job 으로 등록 처리 cacheData.m_strTargetHost = vecProperFhs.front(); m_pCacheJob->Register( cacheData ); } } // Queue Pop 처리 후 // 다음 Query 사용을 대비하여 결과 set clear 처리. m_pQueueCacheRequest->Pop(); rcdb.PgClear(); } // RCDB 연결이 되어 있는 경우.. 종료 처리 if( bDbConnected == true ) { rcdb.PgClear(); rcdb.PgCloseDB(); } return true; }