…·─━☆ 무한코딩세계

1.     목적

 Winsock2 IOCP를 분석한 것입니다. 이 문서는 크게 두 가지를 설명 하고 있습니다. 전자는 일반적인 IOCP에 대해서 설명합니다. 후자는 IOCP란 무엇이고 장단점은 무엇인지 입니다.

 2. IOCP란?

-  PlatformSDK, Network Programming For Microsoft Windows


출처 : http://blog.naver.com/process3/20022353267

프로젝트에서 라이브러리 모듈 목록에 ws2_32.lib파일을 추가하던지,

아니면 아래처럼...

라이브러리 모듈 목록에  추가하는 방법... (VC++ 6기준)

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <winsock2.h>

//  아래 한줄을 서버와 클라이언트 소스에 각각 추가.

#pragma comment(lib, "ws2_32.lib");

#pragma 는 #로 시작하는 전처리구문 지시자 중 컴파일러에 종속적인 명령으로, 
컴파일러에 특정한 옵션 명령을 내리기 위해 사용하는 데요, 

컴파일러에 종속적이기 때문에 컴파일러를 변경했을 경우 실행을 보장하지 못합니다. 

대표적인 #pragma 명령 중에서 once나 pack, comment(lib, *) 등이 있는데 이는 각각 
컴파일러에 

1 once : 해당 소스가 단 한번만 포함되게 한다(여러 번 인클루드 되는 것을 
컴파일러 차원에서 막아줍니다) 

2 pack : 변수 정렬을 인위적으로 변경시킨다(보통은 4바이트로 지정되어 있습니다) 

3 comment(lib, *) : comment로 사용할 수 있는 명령은 여러 개 있는데, 
그중 가장 대표적인 것이 lib 으로, 해당 라이브러리를 링크시켜준다 

과 같은 명령을 내립니다. 



api나 mfc의 구분과 관계없이 해당 컴파일러에서 사용하는 명령이랍니다

MSDN에 있는 내용
code_seg
#pragma code_seg( ["section-name"[,"section-class"] ] )

Specifies a code section where functions are to be allocated. The code_seg pragma specifies the default section for functions. You can, optionally, specify the class as well as the section name. Using #pragma code_seg without a section-name string resets allocation to whatever it was when compilation began.

==========================================================

실행파일에는 코드영역과 데이터영역이 구분되어 있다는 것은 아시지요.
┌───────┐
│                     │
│                     │
│   데이터영역   │
│                     │
│                     │
├───────┤
│                     │
│                     │
│     코드 영역   │
│                     │
│                     │
└───────┘
이런 식의 그림은 아마 어디선가 많이 보셨을겁니다.(그림이 깨지네요.편집할 땐 제대로 보였는데...)
이런 영역을 section이라고 하지요. 위의 설명에 나오는 section이라는 용어가 이것입니다.
실제로 데이터 영역과 코드 영역외에도 exe나 dll에는 여러 section을 포함할 수 있습니다.
이건 dumpbin이라는 툴로 살펴볼 수 있습니다.
제 컴(Windows XP)에서 dumpbin c:\windows\notepadd.exe 를 해 보았더니

2000 .data
2000 .rsrc
7000 .text

이렇게 나오는 군요.

여기서 .data에는 초기화된 데이터가 .rsrc에는 리소스들이, .text에 코드가 들어갑니다.
이러한 .data, .rsrc, .text 등은 일반적으로 정해져 있는 것들입니다.
위 MSDN의 설명에 있는 section-name이라는 것이 바로 .data, .rsrc, .text 등을 뜻하는 겁니다.
즉, #pragma code_seg( .data ) 처럼 사용한다는 거지요.
그리고 Specifies a code section where functions are to be allocated.라는 설명은 
Specifies a section where functions or data or etc. are to be allocated. 이렇게 이해하면 더 나을 듯 하네요.

그런데 이런 정해진 이름말고도 사용자가 새로운 영역을 정할 수 있습니다.

#pragma data_seg("Shared")
DWORD g_dwThreadIdPMRestore = 0;
HWND g_hwnd = NULL;
#pragma data_seg()

이런 식으로 하면 g_dwThreadIdPMRestored와 g_hwnd가 디폴트 데이터 섹션인 .data에 배치되지 않고,
Shared라는 이름으로 만들어진 섹션에 배치되는 것입니다.

=====================================================

#pragma data_seg(...)가 사용되는 곳은 
dll간의 데이터 공유를 위해 사용될 수 있습니다.
그리고 그 외에 어디서 사용되는지는 저도 아직 본 적이 없어서 잘 모르겠습니다.

======================================================

참고로 위의 내용은 "어드밴스 윈도우 NT"를 참고해서 작성했습니다.
이 책은 지금 나오지 않고 이 책의 새 버전이 "Programming Applications for Microsoft Windows"(Jeffrey Richer 저)로
번역판도 나와 있습니다.

주기억장치에서 어떤 작업을 실행하고 있는 동안 그 작업보다 순위가 높은 작업수행이 필요하여

끼어들기 형태로 일을 해야 할 때 이미 실행 중인 프로그램과 데이터를

일시적으로 보조기억장치에 옮기고(swap out), 끼어들기가 끝난 후에 보조기억장치에서

주기억장치로 다시 중단된 작업을 탑재한다(swap in).

이처럼 주기억장치에서 보조기억장치로 프로그램이나 데이터 등을 교환하는 것을 교환이라 한다.

우선 순위를 지닌 작업의 처리를 위한 충분한

기억 장소가 없는 경우에는,

하나 이상의 작업이 주기억장치에서 제거되고 보조기억장치로 교환되어야 한다.

GNU C Library
http://www.gnu.org/software/libc/libc.html

GLIBC 는 GNU C Library 를 뜻합니다.

C언어 자체는 input/ouput , 메모리관리, 문자열 조작 등의 기능이 없습니다.

stdio.h 헤더 파일을 include 해야

printf 함수를 가지고 출력을 하고

scanf 함수를 가지고 입력을 받을 수 있습니다.

malloc , free 함수가 있어야 메모리 관리를 하고

strcmp, strcat 등의 함수가 있어야 문자열을 조작할 수가 있습니다.

이러한 것들을 할 수 있도록 도와주는 C library 중에

GNU 에서 제공하는 GLIBC 가 있는 것 입니다.

uClibc 는 작업환경이 일반 PC 보다 비교적 협소한 임베디드 환경에서 사용할 수 있도록

GLIBC 를 축소한 형태의 library 입니다.

[ 소스 ]

#include "StdAfx.h"
#include
#include "IOCPThreadPool.h"

CIOCPThreadPool::CIOCPThreadPool(void)
{
 m_hStopEvent = NULL;
 m_hCompletionPort = NULL;

 int i = 0;
 for( i = 0 ; i < WORKERTHREAD_COUNT ; i++ )
  m_hWorkerThread[i] = NULL;
 
}

CIOCPThreadPool::~CIOCPThreadPool(void)
{
 CloseHandle(m_hCompletionPort);
 CloseHandle(m_hStopEvent);
}

DWORD CIOCPThreadPool::Create(int nSize)
{
    //  Worker thread 생성
    unsigned int threadID;
    int i;

    //  I/O Completion Port 생성
    m_hCompletionPort = ::CreateIoCompletionPort( INVALID_HANDLE_VALUE, NULL, 0, 0 );
    if( INVALID_HANDLE_VALUE ==  m_hCompletionPort )
            return ERROR_CAN_NOT_COMPLETE;

    for( i = 0 ; i < WORKERTHREAD_COUNT ; i++ )
 {
  m_hWorkerThread[i] = (HANDLE)_beginthreadex( 0, 0, WorkerThread, this, 0, &threadID );
 }

    m_hStopEvent = ::CreateEvent( 0, TRUE, FALSE, 0 );
 
 return 0;
}

DWORD CIOCPThreadPool::InsertQueueItem(DWORD lpData)
{
 PostQueuedCompletionStatus(m_hCompletionPort, 0, (DWORD_PTR)lpData, NULL);
 return 0;
}

unsigned int __stdcall CIOCPThreadPool::WorkerThread(LPVOID lpData)
{
 LPVOID pWorkItem = NULL;
 LPOVERLAPPED lpOverlapped = NULL;
 DWORD numberOfBytes = 0;
 DWORD dwData = 0;
 DWORD dwKey = 0;
 DWORD dwCount = 0;

 CIOCPThreadPool* pThis = (CIOCPThreadPool*)lpData;

    while (1)
    {
  if (WAIT_OBJECT_0 == WaitForSingleObject(pThis->m_hStopEvent, 0))
  {
   break;
  }

        BOOL bSuccess = GetQueuedCompletionStatus(pThis->m_hCompletionPort, &numberOfBytes, &dwKey, &lpOverlapped, INFINITE);

  //AfxTrace(_T("GetQueuedCompletionStatus\n"));
        //
        // Process the work item if there is one.
        //
        if( -1 == dwKey )
        {
   break;
  }
  else
  {
            //Do it...
   dwData = dwKey;
   dwCount++;
  }  
    }

 AfxTrace(TEXT("[%d] Semaphore Use Thread Pool Data|Count[%d, %d] \r\n"), GetCurrentThreadId(), dwData, dwCount);

 return 0;
}
DWORD CIOCPThreadPool::Stop(void)
{
 int i = 0;
 DWORD dwKey = -1;

 SetEvent(m_hStopEvent);
 
 for( i = 0 ; i < WORKERTHREAD_COUNT ; i++ )
  InsertQueueItem(dwKey);

 WaitForMultipleObjects(WORKERTHREAD_COUNT, m_hWorkerThread , TRUE, INFINITE);

 return 0;
}

-------------------------------------------------------------------------------------

 => 앞서 Thread Pool에 대해 간단하게 설명했다....IOCP에서의 쓰레드 풀은

     CreateIoCompletionPort라는 API를 이용하여 운영체제에 IO Port를 쓰겠다는 것을 통보한다.

 => 앞의 쓰레드 풀과 전체적인 의미는 차이가 없다. 다만 앞에는 직접 사용자가 데이터를

     넣고 빼고..동기화도 직접해주는 것이지만 여기서는 OS가 전부 처리해준다는 점만 다르다.

     ( 위의 그림 부분을 OS가 처리 )

 ps. USN 2차에서는 쓰레드풀을 위한 4개의 쓰레드, Read, Write를 위한 2개 쓰레드, 데이터

      처리를 위한 쓰레드1개를 써서 총 7개의 쓰레드가 쓰였다.

=> 앞의 세미포어 대신에 여기서는 IOCP가 대신해주면서 처리한다.

=> 두 방법의 성능차이는 IOCP를 이용한 것이 실제로 세미포어를 이용하여 작성한거 보다

    빠르다.

=> windows2000이상에서는 QueueUserWorker라는 것이 있다. 이는 밑의 그림 처럼 쓰레드 풀

    까지 관리해주는 역활을 한다.

=> 필요성

  - 네트워크 프로그래밍에서 사용자가 늘어날 때마다 쓰레드를 만들 수는 없다.

    어떤 프로그램을 만들 때 몇개의 쓰레드를 만들 것인가? 에 대한 설계가 필요

[소스]

#include "StdAfx.h"
#include "ThreadQueuePool.h"
#include

CThreadQueuePool::CThreadQueuePool(void)
{
 m_hQueueSemaphore = NULL;
 m_hStopEvent = NULL; 
 m_phThreads = NULL;
 m_dwSize = 0;

 InitializeCriticalSection(&CS);

}

CThreadQueuePool::~CThreadQueuePool(void)
{
 if (NULL != m_phThreads)
 {
  delete m_phThreads;
  m_phThreads = NULL;
 }

 DeleteCriticalSection(&CS);

}

DWORD CThreadQueuePool::Create(int nSize)
{
 if (0 >= nSize)
 {
  return ERROR_INVALID_PARAMETER;
 }

 DWORD dwRet = ERROR_SUCCESS;

 m_dwSize = nSize;

 //lInitialCount Value is 0
 m_hQueueSemaphore = CreateSemaphore(NULL, 0, LONG_MAX, NULL);
 if (NULL == m_hQueueSemaphore)
 {
  dwRet = ERROR_CAN_NOT_COMPLETE;
  goto QUIT;
 }
 
 //bManualReset Value is TRUE
 m_hStopEvent = CreateEvent(NULL, TRUE, FALSE, NULL );
 if (NULL == m_hStopEvent)
 {
  dwRet = ERROR_CAN_NOT_COMPLETE;
  goto QUIT;
 }

 //Allocate Thread Handle
 m_phThreads = new HANDLE[nSize];

 for (int i = 0; i < nSize; i++)
 {
  // create thread
  UINT uiThreadId = 0;

  m_phThreads[i] = (HANDLE)_beginthreadex(NULL,
   0,
   CThreadQueuePool::ThreadFunc,
   this,
   0,
   &uiThreadId);

  if (NULL == m_phThreads[i])
  {
   dwRet = ERROR_CAN_NOT_COMPLETE;
   m_dwSize = i;
   StopThread();
   goto QUIT;
  }

 }

QUIT:

 if (ERROR_SUCCESS != dwRet)
 {
  if (NULL != m_hQueueSemaphore)
  {
   CloseHandle(m_hQueueSemaphore);
   m_hQueueSemaphore = NULL;
  }

  if (NULL != m_hStopEvent)
  {
   CloseHandle(m_hStopEvent);
   m_hStopEvent = NULL;
  }
 }

 return dwRet;
}

DWORD CThreadQueuePool::InsertItem(DWORD dwData)
{
 DWORD dwRet = ERROR_SUCCESS;

 {
  EnterCriticalSection(&CS);
  m_aryItemQueue.Add(dwData);
  LeaveCriticalSection(&CS);
 }

 if (FALSE == ReleaseSemaphore(m_hQueueSemaphore, 1, NULL))
 {
  dwRet = ERROR_CAN_NOT_COMPLETE;
  ASSERT(FALSE);
 }

 return dwRet;
}

DWORD CThreadQueuePool::StopThread(void)
{
 if (NULL == m_phThreads)
  return ERROR_SUCCESS;

 DWORD dwReturn = ERROR_SUCCESS;

 SetEvent(m_hStopEvent); 
 dwReturn = WaitForMultipleObjects(m_dwSize, m_phThreads, TRUE, INFINITE);

 if (WAIT_OBJECT_0 == dwReturn)
 {
  AfxTrace(TEXT("All Thread terminate \r\n"));
  delete m_phThreads;
  m_phThreads = NULL;
 }
 else
 {
  ASSERT(FALSE);
 }

 return dwReturn;
}

unsigned int __stdcall CThreadQueuePool::ThreadFunc(void* pParam)
{
 //You must check pParam Value...

 HANDLE hObj[INDEX_END] = {0,};
 DWORD dwResult = 0;
 CThreadQueuePool* pThis = (CThreadQueuePool*)pParam;
 BOOL bGo = TRUE;
 LPVOID lpWorkItem = NULL;
 DWORD dwRet = 0;
 DWORD dwCount = 0;

 hObj[INDEX_QUEUE] = pThis->m_hQueueSemaphore;
 hObj[INDEX_STOP] = pThis->m_hStopEvent;

 while(bGo)
 {
  //This Thread Wait Two Handle
  //bWaitAll value is FALSE
  //dwMilliseconds(WaitTime) value is INFINITE
  dwResult = WaitForMultipleObjects(INDEX_END, hObj, FALSE, INFINITE);

  switch(dwResult)
  {
   // hObj[INDEX_QUEUE] was signaled.
  case WAIT_OBJECT_0 + INDEX_QUEUE:
   //AfxTrace(TEXT("[%d] Semaphore signaled and then Thread Excute \r\n"), GetCurrentThreadId());

   dwCount++;

   {
    EnterCriticalSection(&pThis->CS);
    dwRet = pThis->GetWorkItem();    
    LeaveCriticalSection(&pThis->CS);    
   }
   
   //Doing WorkItem
   
   //Sleep(1000);
   break;

   // hObj[INDEX_STOP] was signaled.
  case WAIT_OBJECT_0 + INDEX_STOP:
   bGo = FALSE;
   break;

  default:
   ASSERT(FALSE);
   break;
  }

 }

 AfxTrace(TEXT("[%d] Semaphore Use Thread Pool Data, Count[%d, %d] \r\n"), GetCurrentThreadId(), dwRet, dwCount);
 
 return 1;
}
DWORD CThreadQueuePool::GetWorkItem(void)
{
 DWORD dwRet = 0;

 if (0 < m_aryItemQueue.GetSize())
 {
  //Synchronization
  dwRet = m_aryItemQueue[0];
  m_aryItemQueue.RemoveAt(0);
 }

 return dwRet;
}

-------------------------------------------------------------------------------------

=> 세마포어를 이용하여 Queue방식으로 쓴다. 세마포어 개념 또한 Count개념이므로...

=> 동기화는 Critical Section으로써 데이터를 넣고 빼고한다.

=> LONG_MAX(소스 참조)만큼 세마포어를 만든다.

=> Insert부분에서 Queue에 Critical Section으로 동기화하고 다른 쓰레드에 쓰라고 통보(ReleaseSemaphore)해준다.

=> 생성한 5개의 쓰레드중에 하나가 꺠어나서 Queue의 데이터를 꺼내서 처리한다. 

출처 : 신경준씨 사외교육 http://blog.naver.com/process3/20052147545

    Critical section는 유저모드에서 작동하는 obj이기 때문이며 같은 프로세스안에서만 사용이

    가능하다.

   - 두개의 Critical section을 돌려도 Mutex보다 빠르다. Critical section은 Count만을 가지고

     있으며 Mutex는 커널까지 내려가서 Critical영역에 대한 관리를 요청하기 때문에 느릴 수 밖에

     없다.

   - 반면에 다른 프로세스에서 다른 프로세스의 자원을 사용할 때는 Mutex를 사용할 수 밖에 없다.

   - 정리하자면 Critical section은 같은 프로세스내에서의 임계영역(Critical 영역)을 관리 한다.

      (위의 머리말 의미와 동일)

-> 코드2듀오 일때 spinCount를 쓰는 것이 5%정도 더 성능이 올라간다.

-> Critical section내에는 sleep를 쓰지 않는 것을 기본으로 하는데 그대신 이벤트처리를

    해도 된다.

[ 세마포어 ]

-> 예제 소스

#include "stdafx.h"
#include
#include "process.h"

#define THREAD_NUM 10
#define INIT_COUNT 0
#define TOTAL_COUNT 10
#define RELEASE_COUNT 1

int g_nIndex = 0;

typedef struct msg_block_tag
{
 HANDLE    hSemaphore; 
} THREAD_ARG;

void PrintTrace()
{
 _tprintf(_T("Thread Exit ID = %d|%0x \n"), g_nIndex, GetThreadId(GetCurrentThread()));
 g_nIndex++;
}

unsigned int WINAPI TestThread(LPVOID lpThreadData);

int _tmain(int argc, _TCHAR* argv[])
{
 unsigned int unThreadId = 0;
 THREAD_ARG  stThreadArg;
 int    i = 0;
 HANDLE   hThread[THREAD_NUM] = {NULL};
 LONG   lPreviousCount = 0;

 stThreadArg.hSemaphore = CreateSemaphore(NULL, INIT_COUNT, TOTAL_COUNT, NULL);

 for( i = 0; i < THREAD_NUM; i++)
 {
  hThread[i] = (HANDLE)_beginthreadex(NULL, 0, TestThread, &stThreadArg, 0, &unThreadId);
 }

 ReleaseSemaphore(stThreadArg.hSemaphore, RELEASE_COUNT, &lPreviousCount );

 WaitForMultipleObjects(THREAD_NUM, hThread, TRUE, INFINITE);
 
 for( i = 0; i < THREAD_NUM; i++)
 {
  CloseHandle(hThread[i]);
  hThread[i] = NULL;
 }

 CloseHandle(stThreadArg.hSemaphore);
 stThreadArg.hSemaphore = NULL;

 return 0;
}

unsigned int WINAPI TestThread(LPVOID lpThreadData)
{

 THREAD_ARG *lpArg = (THREAD_ARG*)lpThreadData;
 DWORD  dwThreadStatus = 0;
 LONG  lPreviousCount = 0;

 while(TRUE)
 {
  dwThreadStatus = WaitForSingleObject(lpArg->hSemaphore, INFINITE);
 
  }
 }

return 0;

}

-> 세마포어의 큰 장점은 숫자를 이용할 수 있어 자기가 원하는 쓰레드 수를 고를 수 있다는 점이다.

    - 소스를 보면 10개의 쓰레드가 모두 wait상태에서 있다가 하나가 벗어나면

      ReleaseSemaphore를  하여 다음 쓰레드에게 알려주게 된다.

    - 세마포어의 ReleaseCount를 1로 하면 Mutex와 똑같이 작동을 한다.

    - 웹 서버개발시 Mutex와 세마포어를 같이 사용할 수도 있다.

      ex) 80개의 쓰레드가 있을 때 Mutex를 쓰면 1개씩 작동을 하므로 나머지 79개가 서로 경쟁

           을 하게 되어 비효율적이다. 그러므로 CPU가 8개라면 세마포어를 이용하여 ReleaseCount

           를 8개로 주고 Mutex를 이용하여 그 8개끼리 경쟁을 하게 하면 더욱더 효율적으로

           구현할 수 있다.( 강의 PPT 15번 참조 )

     - 강의 PPT 15번 내용

---------------------------------------------------------------------------------------- 

Semaphore Throttles (Thundering Herd(놀란 양떼))

 하나의 동기화 객체(크리티컬 섹션 또는 뮤텍스)를 기다리는 스레드가

너무 많은 경우, 세마포어를 하나 더 둬서, 해당 동기화 객체를 기다리는

스레드의 숫자를 줄여주는 것이 성능 향상에 도움이 된다.

while (TRUE) { // Worker loop
   WaitForSingleObject (hThrottleSem, INFINITE);
   WaitForSingleObject (hMutex, INFINITE);
      ... Synchronization code  ...
   ReleaseMutex (hMutex);
   ReleaseSemaphore (hThrottleSem, 1, NULL);
} // End of worker loop

----------------------------------------------------------------------------------------