IOCP를 이용한 Thread Safe Job Queue

IOCP가 사용된 Thread Safe가 지원되는 Job Queue를 공유합니다.
단, queue node의 data size는 sizeof(ULONG_PTR) 이어야 합니다.
IOCP에 대해서는 기회가 된다면 다음에 설명하도록 하겠습니다.

#pragma once // IOCP를 이용한 Thread Safe Job Queue // TYPE : Queue의 DataType // ; (::PostQueuedCompletionStatus(...)의 dwCompletionKey에 들어갈 값으로, ULONG_PTR size여야 한다. // nQueueSize : Queue가 처리할 Size (갯수) // CQueueIOCP Method: // ; DWORD Create(dwWaitThreadNum) // ; VOID Destroy(VOID) // ; VOID StopPush(VOID) // ; BOOL Push(TYPE& data, DWORD dwTimeout = INFINITE) // ; BOOL Pop(TYPE& retData, DWORD dwTimeout = INFINITE) template<typename TYPE, INT nQueueSize> class CQueueIOCP { // Public Method public: CQueueIOCP() { m_hIocp = NULL; m_hSemaphore = NULL; m_bStopPush = TRUE; } ~CQueueIOCP() { Destroy(); } // 초기화, IOCP 생성, 세마포어 초기화 DWORD Create(IN DWORD dwWaitThreadNum) { DWORD dwRtnValue = ERROR_SUCCESS; // Create 하기 전에 일단 Destroy 한다. Destroy(); // Push 가능한 상태로 만든다. m_bStopPush = FALSE; // IOCP Object를 생성한다. m_hIocp = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, dwWaitThreadNum); if (NULL == m_hIocp) { dwRtnValue = ::GetLastError(); MYTRACELOG(TEXT("[ERROR] Fail to CreateIoCompletionPort, CQueueIOCP::Create(), GetLastError=%d"), dwRtnValue); assert(FALSE); goto FINAL; } // Queue Size를 결정하는 Semaphore를 생성한다. m_hSemaphore = ::CreateSemaphore(NULL, nQueueSize, nQueueSize, NULL); if (NULL == m_hSemaphore) { dwRtnValue = ::GetLastError(); MYTRACELOG(TEXT("[ERROR] Fail to CreateSemaphore, CQueueIOCP::Create, GetLastError=%d"), dwRtnValue); assert(FALSE); goto FINAL; } FINAL: return dwRtnValue; } // 자원정리, IOCP 객체 및 세마포어 정리 void Destroy(VOID) { // Push를 중단한다. StopPush(); if (NULL != m_hIocp) { ::CloseHandle(m_hIocp); m_hIocp = NULL; } if (NULL != m_hSemaphore) { ::CloseHandle(m_hSemaphore); m_hSemaphore = NULL; } } // Push를 중단한다. // 주로, Destory를 호출하기 전에, 호출하여, Destroy(...) 호출 이후에 // 다른 Thread에서 Push하는 것을 방지한다. VOID StopPush(VOID) { m_bStopPush = TRUE; } // Push를 재개한다. VOID StartPush(VOID) { m_bStopPush = FALSE; } // Queue에 Data를 Push 한다. // dwTimeout에 값이 있으면, blocking 모드이다. // ; 큐에 빈 자리가 없으면 타임아웃 시간까지 블러킹, 자리 생기면 넣고 리턴 // dwTimeout이 0 이면, non-blocking 모드이다. // ; 큐에 자리가 있으면 넣고 자리 없으면 바로 FALSE라는 결과와 함께 리턴 // 만약, StopPush 중이라면, TRUE가 리턴된다. // 만약, Handle이 Closed(즉, Destroy() 호출)되었다면, TRUE가 리턴된다. BOOL Push(IN TYPE& data, IN DWORD dwTimeout = INFINITE) { DWORD dwRet = 0; if (TRUE == m_bStopPush) { // Stop Push 인가? // 이때는 Push가 성공한 것으로 리턴한다. return TRUE; } if ((NULL == m_hSemaphore) || (NULL == m_hIocp)) { assert(FALSE); return FALSE; } // Semaphore를 Wait한다. dwRet = ::WaitForSingleObject(m_hSemaphore, dwTimeout); if (WAIT_OBJECT_0 == dwRet) { // 제대로 Wait 되었다. // do nothing,... 하고 아래의 PostQueuedCompletionStatus(...)를 실행한다. } else { // Timeout 되었을 것이다. #ifdef _DEBUG MYTRACELOG(TEXT("[ERROR] Fail to WaitForSingleObject in CQueueIOCP Push, dwRet=%d"), dwRet); #endif assert(FALSE); return FALSE; } if (TRUE == m_bStopPush) { // (Semaphore waiting 중에 StopPush(...)가 호출될 수 있으므로, 여기에서도 // StopPush를 체크한다.) // Stop Push 인가? // 이때는 Push가 성공한 것으로 리턴한다. return TRUE; } // IOCP Queue에 추가한다. if (FALSE == ::PostQueuedCompletionStatus(m_hIocp, 0, (ULONG_PTR)data, NULL)) { dwRet = ::GetLastError(); // Handle이 Closed 되었음. if (ERROR_INVALID_HANDLE == dwRet) { return TRUE; } // Queue에 넣는것이 실패했다. #ifdef _DEBUG MYTRACELOG(TEXT("[ERROR] Fail to PostQueuedCompletionStatus in CQueueIOCP Push, GetLastError=%d"), dwRet); #endif assert(FALSE); return FALSE; } // IOCP Queue에 추가 성공 return TRUE; } // Queue에 Data를 Pop한다. // dwTimeout에 값이 있으면, blocking 모드이다. // ; 큐에 아이템이 없으면 타임아웃 시간까지 블러킹, 아이템 있으면 리턴 // dwTimeout에 값이 없으면, non-blocking 모드이다. // ; 큐에 아이템이 있으면 가지고 오고 TRUE 리턴, 아이템 없으면 바로 FALSE 와 함께 리턴 // 만약, Handle이 Closed(즉, Destroy() 호출)되었다면, TRUE가 리턴된다. (pbClosed에 TRUE가 들어간다.) BOOL Pop(OUT TYPE& retData, OUT LPBOOL pbClosed, IN DWORD dwTimeout = INFINITE) { BOOL bRet = FALSE; DWORD dwTemp = 0; LPOVERLAPPED pTempOverlapped = NULL; TYPE PopingItem; if (NULL == pbClosed) { assert(FALSE); return FALSE; } // 일단 Close 아님. *pbClosed = FALSE; // IOCP Queue로 부터 값을 가져온다. bRet = ::GetQueuedCompletionStatus(m_hIocp, &dwTemp, ((PULONG_PTR)(&PopingItem)), &pTempOverlapped, dwTimeout); if (TRUE == bRet) { bRet = ::ReleaseSemaphore(m_hSemaphore, 1, NULL); if (FALSE == bRet) { #ifdef _DEBUG MYTRACELOG(TEXT("[ERROR] Fail to ReleaseSemaphore in CQueueIOCP Pop, GetLastError=%d"), ::GetLastError()); #endif assert(FALSE); } // 값을 전달한다. retData = PopingItem; } else { dwTemp = ::GetLastError(); // Handle이 Closed 되었음. if (ERROR_INVALID_HANDLE == dwTemp) { *pbClosed = TRUE; return TRUE; } #ifdef _DEBUG MYTRACELOG(TEXT("[ERROR] Fail to GetQueuedCompletionStatus in CQueueIOCP Pop, GetLastError=%d"), dwTemp); #endif assert(FALSE); return FALSE; } // 성공 return TRUE; } // Private Member private: HANDLE m_hIocp; HANDLE m_hSemaphore; BOOL m_bStopPush; }; <SPAN id=tx_marker_caret></SPAN>