windows IOCP 기초

IOCP 개념

---

싱글 쓰레드로는 부족해

이전 게시물까지 이야기 했던 통지모델들은 싱글쓰레드 멀티플렉싱을 위한 확장처럼 보였다. 하지만 싱글쓰레드 스마트폰도 보기 힘든 2015년 현재, 굳이 하이엔드 스펙을 갖춘 서버에서 싱글쓰레드를 써야할까? 이전 멀티쓰레드형 서버의 문제는 컨텍스트 스위칭 비용이었다. 하지만 CPU개수만큼만 쓰레드를 사용한다면, 컨텍스트 스위칭 문제는 크게 문제가 되지 않는다. 그러니까 우리는 딱 CPU개수만큼만 쓰레드를 쓰는 서버를 만들고 싶다. 이 막연한 희망사항에 긍정적으로 대답해주는 것이 바로 윈도우의 IOCP이다.

어떻게 구현할 것인가?

수많은 I/O 요청속에서 딱 CPU 개수만큼만 쓰레드를 사용하여 처리한다는게 말이 쉽지 땅파면 나오나? 어떻게 만들 것인가? 일단 생각나는 대로 방법을 도마위에 올려보자. 우선은 I/O 장치와 통지하는 객체를 연결한다. 그리고 여러 I/O 장치로부터 나오는 동시다발적 I/O요청을 효율적으로 처리하기 위해서 Overlapped I/O에서 사용한 비동기 통지방식을 사용한다. 여러 쓰레드를 효율적이고 민감하게 관리할 수 있어야 하니까 쓰레드 풀을 만든다. I/O가 종료된 이후에 발생한 완료 통지를 쓰레드 풀에 의해 통제된 쓰레드에게 하나씩 맡긴다면? 이제 좀 윤곽이 보이는 것 같다. 쓰레드 활용이 제한적이니 우선 완료된 I/O들의 정보를 종료 순서대로 큐에 담는다. 쓰레드가 하는 일이 끝나면 큐에서 완료된 I/O를 새로 하나 받아온다. 쓰레드에서 완료 I/O 정보에 따라서 적절한 처리를 실행한다. 

만약 후처리하는 쓰레드가 Blocking 되어서 CPU를 제대로 안쓰고 있다면, 리소스 낭비다. 이런 쓰레드는 잠시 고이접어두고 새로운 쓰레드가 활성화 될 수 있게하자. 이때 주의해야 하는건 전체 활성화 된 쓰레드가 반드시 CPU 개수만큼이어야 된다는 점이다. 쓰레드 풀은 그점을 주의하여 쓰레드를 활성화시킨다. 

이렇게 만들어진 구조의 도움을 받아서 사용자(프로그래머)는 디바이스를 통지객체에 연결시킨다음, 일을 처리할 쓰레드들을 통지객체에 묶어서 대기시킨다. 그리고 Overapped I/O 방식을 그대로 적용한채 I/O 정보를 잘 구조화해서 Overapped 구조체나 전달 가능한 인자를 통해 통지객체에 등록한다. 그리고 쓰레드들은 완료시 어떻게 처리할지 명시해주면, I/O가 완료될때마다 노는 쓰레드가 알아서 척척 일사분란하게 I/O 작업을 처리할 것이다!

여기까지의 상상력을 직접 구현하려면 몇 가지 자료구조의 도움을 받아야 할 거다. 완료된 I/O의 대기열. 완료I/O으의 정보를 받아서 실행할 수 있는 쓰레드들의 대기열. 현재 활성화되어 처리중인 쓰레드들의 리스트. 잠시 작업을 위해 CPU사용을 중단하고 있는 쓰레드들의 리스트. 여기에 IOCP가 처리할 I/O 디바이스들의 리스트를 추가하면 IOCP의 동작에 핵심적인 자료구조들을 다 나열 하였다.

IOCP 동작 원리

---

언급한 자료구조들을 실제로 IOCP가 어떻게 다루는지, 그리고 실제 유저가 그것을 어떻게 사용할 수 있는지만 잘 이해한다면, 이 통지 방식의 동작 원리에 대해서 잘 이해할 수 있다고 믿는다. 

디바이스 리스트

I/O 처리를 하려면 우선 I/O 디바이스(소켓, FD) IOCP에 등록을 해야한다. CreateIoCompletionPort 함수를 통해서 디바이스와 CompletionPort(이하 CP)를 바인딩한다. 그런데 CraeteIoCompletionPort 함수의 이름을 다시 한번 생각해보면 디바이스에 바인딩하는 것과는 조금 의미가 다른것 같다. 실제로 이 함수는 두가지 기능을 한다. 첫 번째는 통지모델인 CP를 생성하고 그 핸들을 반환하는 것. 두 번째는 위에서 말한것처럼 디바이스와 CP를 바인딩하는것. 하나의 함수로 서로다른 두 가지 일을 하니, 가능하다면 다른 이름의 함수로 싸서 사용하는게 좋을 듯 하다.

HANDLE WINAPI CreateIoCompletionPort(
  _In_      HANDLE FileHandle,
  _In_opt_  HANDLE ExistingCompletionPort,
  _In_      ULONG_PTR CompletionKey,
  _In_      DWORD NumberOfConcurrentThreads
);

함수의 signature는 위와 같다. 첫번째 인자에 I/O 디바이스의 핸들을, 두 번째 인자에 CP의 핸들을 넘겨서 사용한다. 세 번째 인자인 Key에는 유저가 원하는 데이터를 넘겨서, I/O 통지때 같이 받을 수 있다. 디바이스(소켓) 정보가 담긴 객체(ex: 세션)를 넘겨서 사용하면 편리하다. 네번째 인자는 동시에 작동할 쓰레드의 최대 수를 정하는데 사용된다. 우리의 목표는 CPU개수만큼 쓰레드를 쓰는 것이므로 여기에 CPU 개수를 집어넣으면 OK다. 물론 MS도 많은 이들이 같은 목표를 공유한다는 사실을 잘 알기 때문에 0만 넣어도 CPU개수로 척척 맞춰준다. 

HANDLE hPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);

CP를 처음 만들때는 위와 같이 하면 된다. CP의 핸들이 반환된다. 이 핸들은 앞으로 종종 사용되니 잘 보관한다.

HANDLE port = CreateIoCompletionPort(socket, hPort, (ULONG_PTR)session, 0);

디바이스와 연결할 떄는 위와 같이 사용한다. Key는 디바이스에 고유한 정보를 넣어주면 편리하게 쓸 수 있다.

I/O Completion Queue

완료된 I/O들의 정보가 저장되는 대기큐. I/O 완료시 말고도 유저가 직접 PostQueuedCompletionStatus 함수(이하 PQCS)를 호출하여 정보를 직접 집어 넣을 수도 있다. 이 함수를 보면 어떤 정보가 전달되는지 잘 알 수 있다. 

BOOL WINAPI PostQueuedCompletionStatus(
  _In_      HANDLE CompletionPort,
  _In_      DWORD dwNumberOfBytesTransferred,
  _In_      ULONG_PTR dwCompletionKey,
  _In_opt_  LPOVERLAPPED lpOverlapped
);

통지할 CP를 인자를 결정한다. I/O의 경우 이미 I/O 디바이스와 바인딩된 CP에 전달 될 것이다. 그리고 전송된 바이트의 크기와 Key를 넘길 수 있다. I/O의 경우 디바이스 바인딩할 때 미리 넣어둔 그 Key가 전달된다. 마지막으로 버퍼가 담긴 Overlapped 구조체를 넘겨줄 수 있다. 기존 Overlapped I/O 함수에서 전달하는 Overlapped 구조체가 이에 해당한다. 위에서 언급한 정보들을 활용하면 완료된 I/O에 대해서 충분히 잘 처리할 수 있다. 이렇게 전달한 정보들은 I/O가 완료되면 Completion Queue 에 저장되어 쓰레드가 나타나 자신을 처리해주기만을 기다리게 된다.

Waiting Thread Queue

쓰레드들이라고 노는건 아니다. 다만 바쁘거나 나설 자리가 없을 뿐 (우리가 활성화할 쓰레드 개수를 제한한 것을 잊지말자). 사용자는 쓰레드를 만들어 대기상태로 만들어야 한다. 이때 사용하는 함수가 GetQueuedCompletionStatus (이하 GQCS) 이다. 이름만 보면 바로 I/O 정보를 뽑아올 것 같지만, 우리가 앞에서 말했던것을 곰곰히 생각해보면 왜 그러면 안되는지를 알 수 있을 것이다. 정보를 받아 올 수 있는 상황이 될때까지, 즉 자신이 올라갈 빈 CPU공간이 있고(허용된 활성화 쓰레드 > 현재 활성화된 쓰레드) && 자신을 필요로 하는 완료된 I/O가 있는 경우에 비로소 이 정보를 받아 올 수 있다. 이런 조건이 맞아 떨어지지 않으면, 설정한 timeout까지 이 쓰레드는 대기상태가 된다. 

Release Thread List

대기상태가 풀리고 쓰레드가 활성화 됬다는 것은 (timeout 제외), 드디어 I/O를 처리할 수 있다는 뜻이다. I/O장치 또는 PQCS로 보낸 정보를 GetQueuedCompletionStatus 함수를 통해서 받아온다.

BOOL WINAPI GetQueuedCompletionStatus(
  _In_   HANDLE CompletionPort,
  _Out_  LPDWORD lpNumberOfBytes,
  _Out_  PULONG_PTR lpCompletionKey,
  _Out_  LPOVERLAPPED *lpOverlapped,
  _In_   DWORD dwMilliseconds
);

생긴것을 보면 딱 Post와 짝이다. 마지막 인자는 미리 언급한 timeout 설정 시간이다. 마음껏 I/O 처리를 하고난 다음 다시 GetQueuedCompletionStatus를 호출하면 다시 대기상태로 들어간다. 이 함수를 다시 호출하지 않으면 다른 쓰레드가 동작할 수 없으니 조심해야한다. 

Paused Thread List

엄밀히 말하자면 Released Thread Queue의 마지막 문장은 잘못되었다. Released List에 있는 쓰레드, 즉 활성화된 쓰레드 중에서 Blocking 상태에 빠진녀석들을 제대로 처리하지 않는다면 비효율적이다. GQCS를 호출하지 않아도 쓰레드가 Blocking상태에 빠지면, IOCP는 똑똑하게 그것을 감지하여 이 Paused Thread List에 집어넣는다. 그러면 가용 쓰레드 공간이 늘어나서 CPU가 필요한 다른 쓰레드에게 리소스를 양도할 수 있는 것이다. 

IOCP 사용 예

---

쓰레드에서 

while (true)
{
    DWORD dwTransferred = 0;
    OverlappedIOContext* context = nullptr;
    ClientSession* asCompletionKey = nullptr;  //키로 세션을 활용

    int ret = GetQueuedCompletionStatus( 
        hComletionPort , &dwTransferred, 
        (PULONG_PTR)&asCompletionKey, (LPOVERLAPPED*)&context, GQCS_TIMEOUT );
    DWORD errorCode = GetLastError();

    //time out처리
    if (ret == 0 &&  errorCode == WAIT_TIMEOUT)
        continue;
    //기타 에러 처리
        ...

        //overlapped 구조체에 I/O 타입을 전달하는 방법
    switch (context->mIoType)
    {
        //각 I/O에 대응하는 완료함수를 불러서 처리한다.
    case IO_SEND:
        completionOk = SendCompletion(asCompletionKey, context, dwTransferred);
        break;

    case IO_RECV:
        completionOk = ReceiveCompletion(asCompletionKey, context, dwTransferred);
        break;
    }
}

bool ReceiveCompletion(const ClientSession* client, 
                       OverlappedIOContext* context, 
                       DWORD dwTransferred)
{
    /// echo back 처리 client->PostSend()사용.
    bool result = true;
    if( !client->PostSend( context->mBuffer , dwTransferred ) )
    {
        printf_s( "PostSend error: %d\n" , GetLastError() );
        delete context;
        return false;
    }

    delete context;
    return client->PostRecv();
}

세션에서 

bool ClientSession::PostRecv() const
{
    if (!IsConnected())
        return false;

        //커스텀 Overlapped 객체 사용!
    OverlappedIOContext* recvContext = new OverlappedIOContext(this, IO_RECV);

    DWORD recvbytes = 0;
    DWORD flags = 0;
    recvContext->mWsaBuf.buf = recvContext->mBuffer;
    recvContext->mWsaBuf.len = BUFSIZE;

        //WSARecv를 사용!
    DWORD ret = WSARecv( 
        mSocket , &( recvContext->mWsaBuf ) , 1 , 
        &recvbytes , &flags , 
        ( LPWSAOVERLAPPED )recvContext , NULL );

    //에러 처리
    return true;
}