[TCP/IP 소켓 프로그래밍] (16) 비동기 모델 Overlapped IO

출처 : 열혈 TCP/IP 소켓 프로그래밍 윤성우 저

 

비동기 Notification IO

IO 완료에 대한 알림이 비동기적으로 처리되는 것

 

IO(입출력)의 중첩(Overlapped)

Overlapped IO 모델 / 이미지 출처 : TCP/IP 소켓프로그래밍 (윤성우 저)

 하나의 쓰레드 내에서 동시에 둘 이상의 영역으로 데이터를 전송함으로 인해서 입출력이 중첩되는 상황을 가리켜 'IO 중첨' 이라 한다. 이러한 일이 가능하려면 기본적으로 IO 입출력 함수는 논블로킹 (바로 반환하는) 함수여야 된다. 위 그림에서 각 소켓 B, C, D을 목적으로하는 출력버퍼를 통해 데이터를 중첩하여 보내는 모델이 Overlapped IO 모델이자 비동기 IO 모델이라 할 수 있다.

 

Overlapped IO 모델의 중점

IO가 비동기적으로 일어나는 점보다 중요한 것은 IO가 완료된 상황을 확인하는 방법에 있다. 비동기적 모델의 입출력 결과를 나중에 확인해야하기 때문이다.

 

Overlapped IO 소켓 생성

일반적인 소켓을 사용하는 것이 아니라 비동기적 행동이 가능한 소켓을 생성한다.

SOCKET WSASocket(int af, int type, int protocol, LPWSAPROTOCOL_INFO lpProtocolInfo, GROUP g, DWORD dwFlags);

 lpProtocolInfo: 소켓 특성 정보를 전달, WSAPROTOCOL_INFO 구조체 이용, 없다면 NULL

g: 함수 확장을 위해 예약되어 있는 매개변수, 혹은 없다면 0 전달

dwFlags: 소켓의 특성 정보 전달

 

Overlapped IO가 가능한 논블로킹 모드의 소켓

WSASocket(PF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);

마지막 특성 정보에 WSA_FLAG_OVERLAPPED를 설정한다.

 

Overlapped IO를 위한 WSASend 함수

int WSAAPI WSASend(
  [in]  SOCKET                             s,
  [in]  LPWSABUF                           lpBuffers,
  [in]  DWORD                              dwBufferCount,
  [out] LPDWORD                            lpNumberOfBytesSent,
  [in]  DWORD                              dwFlags,
  [in]  LPWSAOVERLAPPED                    lpOverlapped,
  [in]  LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine
);

• lpBuffers : 전송할 데이터를 담은 WSABUF 구조체 배열의 주소값
• lpNumberOfBytesSend: 전송된 바이트 수가 저장될 주소 값
• lpOverlapped: WSAOVERLAPPED 구조체 변수의 주소 값. Event 오브젝트가 포함되어 데이터 전송의 완료를 확인한다.
• lpCompletionRoutine: Completion Routine 이라는 함수의 주소값

 

WSABUF 구조체

실제 전송할 데이터의 주소와 데이터 크기를 담는다.

typedef struct __WSABUF
{
    u_long len; // 전송할 데이터의 크기
    char FAR* buf; // 버퍼의 주소 값
} WSABUF, *LPWSABUF;

 

WSASend를 호출하는 로직

전송할 데이터와  Event 객체를 할당한 overlapped 객체를 WSASend 함수의 인자로 전달하여 사용한다.

WSAEVENT event;
WSAOVERLAPPED overlapped = {};
WSABUF dataBuf;
char buf[BUF_SIZE] = {"전송할 데이터"};
int recvBytes = 0;

// ...
event = WSACreateEvent();
ovelapped.hEvent = event;
dataBuf.len = sizeof(buf);
dataBuf.buf = buf;
WSASend(hSocket, &dataBuf, 1, &recvBytes, 0, &overlapped, NULL);
// ...

overlapped 객체는 전송완료시 event의 커널 오브젝트에 signal 상태로 기록된다.

 

WSOVERLAPPED 구조체

typedef struct _WSAOVERLAPPED
{
   DWORD Internal;
   DWORD ...;
   WSAEVENT hEvent; // 이것만 사용된다.
} WSAOVERLAPPED, *LPWSAOVERLAPPED;

 

Overlapped IO로 동작하는 WSASend 사용시 lpOverlapped인자는 반드시 NULL이 아닌 구조체이어야 하고 또 다른 전송을 하고자할 때, 다른 WAOVERLAPPED 구조체를 생성하여 전달해야한다. (1 전송 - 1 WSOVERLAPPED 구조체)

 

전송 완료 시점에 대하여

WSASend 함수는 바로 반환하는 함수이다. 만약 전송하는 데이터 양이 적다면 전송이 바로 완료되어 함수가 반환될 수 도 있다. 이런 경우 함수는 0을 반환하고 lpBuffers 인자에 전송된 바이트 수가 기록된다.

하지만, 아직 전송 중 인 경우 WSASend는 SOCKET_ERROR를 반환하게 된다. 실제 전송 중인 상황과 전송 실패 상황인지 판별하기 위해 WSAGetLastError 함수 호출을 통해 WS_IO_PENDING인지 판별하도록 한다.

 

송수신 후 결과를 알기 위한 WSAGetOverlappedResult 함수

BOOL WSAGtOverlappedResult(SOCKET s, LPWSAOVERLAPPED lpOverlapped, LPDWORD lpcbTransfer,
							BOLL fWait, LPDWORD lpdwFlags);

• 성공 시 TRUE, 실패 시 FALSE
• s: Overlapped IO가 진행된 소켓 핸들
• lpOverlapped: 등록된 WSAOVERLAPPED 구조체 변수의 주소
• lpcbTransfer: 실제 송수신된 바이트 크기를 저장할 변수의 주소
• fWait: 여전히 IO가 진행중이라면, TRUE 전달 시 대기하고 FALSE 전달시 함수를 반환한다.
• lpdwFlags: WSARecv함수가 호출될 경우 부수적인 메세지 정보 (OOB와 같은)를 얻기 위해 사용한다.

Overlapped IO를 위한 WSARecv 함수

int WSAAPI WSARecv(
  [in]      SOCKET                             s,
  [in, out] LPWSABUF                           lpBuffers,
  [in]      DWORD                              dwBufferCount,
  [out]     LPDWORD                            lpNumberOfBytesRecvd,
  [in, out] LPDWORD                            lpFlags,
  [in]      LPWSAOVERLAPPED                    lpOverlapped,
  [in]      LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine
);

• lpBuffers : 수신할 데이터를 담은 WSABUF 구조체 배열의 주소값
• lpNumberOfBytesSend: 수신된 바이트 수가 저장될 주소값
• lpOverlapped: WSAOVERLAPPED 구조체 변수의 주소값.
• lpCompletionRoutine: Completion Routine 이라는 함수의 주소값.

 

Overlapped IO에서 입출력 완료 확인 예제

입출력 완료와 결과를 확인 하는 방법

1. WSAOVERLAPPED 구조체의 Event 객체가 signal 상태인지 확인
2. CompletionRoutine 함수의 호출

1. Event 객체를 이용

Signaled 상태로 Event 객체가 반환되면 WSAGetOverlappedResult 함수로 결과를 확인한다.

 

수신측 코드

#include "../network_header.h"
#include "../ErrorHandling.h"
#include <WinSock2.h>

#define BUF_SIZE 1024

int main(int argc, char* argv[])
{
    // 윈속 라이브러리 초기화
    WSADATA wsaData;
    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
        ErrorHandling("WSAStartup() error");

    // Overlapped IO 서버 소켓 생성
    SOCKADDR_IN lisnAdr;
    SOCKET hLisnSock = WSASocket(PF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
    memset(&lisnAdr, 0, sizeof(lisnAdr));
    lisnAdr.sin_family = AF_INET;
    lisnAdr.sin_addr.s_addr = htonl(INADDR_ANY);
    lisnAdr.sin_port = htons(PORT);

    if (bind(hLisnSock, (SOCKADDR*)&lisnAdr, sizeof(lisnAdr)) == SOCKET_ERROR)
        ErrorHandling("bind() error");
    if (listen(hLisnSock, 5) == SOCKET_ERROR)
        ErrorHandling("listen() error");

    // 소켓 연결
    SOCKADDR_IN recvAdr;
    int recvAdrSz = sizeof(recvAdr);
    SOCKET hRecvSock = accept(hLisnSock, (SOCKADDR*)&recvAdr, &recvAdrSz);

    // 수신 버퍼
    WSABUF dataBuf;
    char buf[BUF_SIZE] = {};
    DWORD recvBytes = 0, flags = 0;

    // WSAOVERLAPPED 구조체에 등록할 이벤트
    WSAEVENT sockEvent = WSACreateEvent();
    WSAOVERLAPPED overlapped = {};
    overlapped.hEvent = sockEvent;
    dataBuf.len = BUF_SIZE;
    dataBuf.buf = buf;

    // Overlapped IO 방식의 데이터 수신
    if (WSARecv(hRecvSock, &dataBuf, 1, &recvBytes, &flags, &overlapped, NULL) == SOCKET_ERROR)
    {
        // 아직 수신중인 경우
        if (WSAGetLastError() == WSA_IO_PENDING)
        {
            puts("Background data is receiving...");
            // 전송 완료가 될 때까지 기다림
            WSAWaitForMultipleEvents(1, &sockEvent, TRUE, WSA_INFINITE, FALSE);
            // 전송 완료 후 기록된 결과
            WSAGetOverlappedResult(hRecvSock, &overlapped, &recvBytes, FALSE, NULL);
        }
        else
        {
            ErrorHandling("WSARecv() error");
        }
    }

    printf("Received message : %s \n", buf);
    WSACloseEvent(sockEvent);
    closesocket(hRecvSock);
    closesocket(hLisnSock);
    // 윈속 라이브러리 해제
    WSACleanup();
    return 0;
}

송신측 코드

#include "../network_header.h"
#include "../ErrorHandling.h"
#include <WinSock2.h>

int main(int argc, char* argv[])
{
    // 윈속 라이브러리 초기화
    WSADATA wsaData;
    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
        ErrorHandling("WSAStartup() error");

    // Overlapped IO 소켓 생성
    SOCKADDR_IN sendAdr;
    SOCKET hSocket = WSASocket(PF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
    memset(&sendAdr, 0, sizeof(sendAdr));
    sendAdr.sin_family = AF_INET;
    sendAdr.sin_addr.s_addr = inet_addr(DEFAULT_NETWORK);
    sendAdr.sin_port = htons(PORT);

    if (connect(hSocket, (SOCKADDR*)&sendAdr, sizeof(sendAdr)) == SOCKET_ERROR)
        ErrorHandling("connect() error");

    // 전송할 데이터
    WSABUF dataBuf;
    char msg[] = "Network is Computer";
    DWORD sendBytes = 0;

    // WSAOVERLAPPED 구조체에 등록할 이벤트
    WSAEVENT sockEvent = WSACreateEvent();
    WSAOVERLAPPED overlapped = {};
    overlapped.hEvent = sockEvent;
    dataBuf.len = strlen(msg) + 1;
    dataBuf.buf = msg;

    // Overlapped IO 방식의 데이터 전송
    if (WSASend(hSocket, &dataBuf, 1, &sendBytes, 0, &overlapped, NULL) == SOCKET_ERROR)
    {
        // 아직 전송중인 경우
        if (WSAGetLastError() == WSA_IO_PENDING)
        {
            puts("Background data is sending...");
            // 전송 완료가 될 때까지 기다림
            WSAWaitForMultipleEvents(1, &sockEvent, TRUE, WSA_INFINITE, FALSE);
            // 전송 완료 후 기록된 결과
            WSAGetOverlappedResult(hSocket, &overlapped, &sendBytes, FALSE, NULL);
        }
        else
        {
            ErrorHandling("WSASend() error");
        }
    }

    printf("Send data size : %d\n", sendBytes);
    WSACloseEvent(sockEvent);
    closesocket(hSocket);

    // 윈속 라이브러리 해제
    WSACleanup();
    return 0;
}

2. Completion Routine의 사용

 

Pending 이 완료가 된 이후 호출할 콜백 함수를 등록하는 것이다. 이러한 콜백 함수 호출 시기는 중요한 계산 중에 갑자기 호출되면 프로그램의 흐름이 끊길 수 있다. 따라서 IO를 요청한 쓰레드가 "alertable wait" 상태에 놓여있을 때만 Completion Routine 함수가 호출될 수 있도록 한다.

 

"alertable wait" 상태는 특별한 일을 하지 않고 운영체제의 메세지를 수신하는 것을 기다리는 쓰레드의 상태를 의미한다. 다음 함수들을 통해 쓰레드 상태를 alertable wait = TRUE로 설정할 수 있다.

WaitForSingleObjectEx
WaitForMultipleObjectsEx
WSAWaitForMultipleEvents
SleepEx

위 함수들은 Ex가 붙지 않은 함수들과 똑같은 동작을 하나 추가된 마지막 매개변수에 Boolean 값을 전달하여 alertable wait를 설정할 수 있다. 이 값이 TRUE로 설정되어 호출 시점에 IO가 완료된 것이 있다면 Completion Routine를 실행하고 해당 함수는 WAIT_IO_COMPLETION를 반환한다.

 

수신측 코드

#include "../network_header.h"
#include "../ErrorHandling.h"
#include <WinSock2.h>

#define BUF_SIZE 1024
// 콜백 함수
void CALLBACK CompRoutine(DWORD, DWORD, LPWSAOVERLAPPED, DWORD);

// 수신 버퍼
WSABUF dataBuf;
char buf[BUF_SIZE] = {};
DWORD recvBytes = 0, flags = 0;

int main(int argc, char* argv[])
{
    // 윈속 라이브러리 초기화
    WSADATA wsaData;
    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
        ErrorHandling("WSAStartup() error");

    // Overlapped IO 서버 소켓 생성
    SOCKADDR_IN lisnAdr;
    SOCKET hLisnSock = WSASocket(PF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
    memset(&lisnAdr, 0, sizeof(lisnAdr));
    lisnAdr.sin_family = AF_INET;
    lisnAdr.sin_addr.s_addr = htonl(INADDR_ANY);
    lisnAdr.sin_port = htons(PORT);

    if (bind(hLisnSock, (SOCKADDR*)&lisnAdr, sizeof(lisnAdr)) == SOCKET_ERROR)
        ErrorHandling("bind() error");
    if (listen(hLisnSock, 5) == SOCKET_ERROR)
        ErrorHandling("listen() error");

    // 소켓 연결
    SOCKADDR_IN recvAdr;
    int recvAdrSz = sizeof(recvAdr);
    SOCKET hRecvSock = accept(hLisnSock, (SOCKADDR*)&recvAdr, &recvAdrSz);

    // WSAOVERLAPPED 구조체에 등록할 이벤트
    WSAEVENT sockEvent = WSACreateEvent();
    WSAOVERLAPPED overlapped = {};
    overlapped.hEvent = sockEvent;
    dataBuf.len = BUF_SIZE;
    dataBuf.buf = buf;

    // Overlapped IO 방식의 데이터 수신
    if (WSARecv(hRecvSock, &dataBuf, 1, &recvBytes, &flags, &overlapped, CompRoutine) == SOCKET_ERROR)
    {
        // 아직 수신중인 경우
        if (WSAGetLastError() == WSA_IO_PENDING)
        {
            puts("Background data is receiving...");
            //// 전송 완료가 될 때까지 기다림
            //WSAWaitForMultipleEvents(1, &sockEvent, TRUE, WSA_INFINITE, FALSE);
            //// 전송 완료 후 기록된 결과
            //WSAGetOverlappedResult(hRecvSock, &overlapped, &recvBytes, FALSE, NULL);
        }
        else
        {
            ErrorHandling("WSARecv() error");
        }
    }

    int idx = WSAWaitForMultipleEvents(1, &sockEvent, FALSE, WSA_INFINITE, TRUE);
    if (idx == WAIT_IO_COMPLETION)
        puts("Overlapped I/O Completed");
    else
        ErrorHandling("WSARecv Error()");

    //printf("Received message : %s \n", buf);
    WSACloseEvent(sockEvent);
    closesocket(hRecvSock);
    closesocket(hLisnSock);
    // 윈속 라이브러리 해제
    WSACleanup();
    return 0;
}

void CALLBACK CompRoutine(DWORD dwError, DWORD szRecvBytes, LPWSAOVERLAPPED lpOverlapped, DWORD flags)
{
    if (dwError != 0)
    {
        ErrorHandling("CompRoutine error");
    }
    else {
        szRecvBytes = szRecvBytes;
        printf("Received message : %s\n", buf);
    }
}