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

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

 

동기화된 입출력 함수의 동작 방식

이미지 출처 : 열혈 TCP/IP 소켓프로그래밍 (윤성우 저)

전송 : send가 호출되는 순간 출력 버퍼로 데이터 쓰기가 시작되고 데이터 쓰기가 완료되면 send 함수가 반환된다.

수신 : recv가 호출되는 순간 입력 버퍼로 데이터를 읽기가 시작되고 데이터 읽기가 완료되면 recv 함수가 반환된다.

 

select 함수는 대표적인 동기적 Notification 모델의 예로 관찰대상의 파일 디스크립터 중 하나라도 변화가 있을 시 반환하는 함수이기 때문이다. 즉, IO가 필요한 상황이 알림이 되는 시점이 함수가 반환되는 시점과 일치한다.

 

동기화된 "입출력"의 단점

입출력이 진행되는 동안 호출된 함수가 반환을 하지 않으니, 다른 일을 할 수가 없다.

 

비동기적 입출력 함수의 동작 방식

이미지 출처 : 열혈 TCP/IP 소켓프로그래밍 (윤성우 저)

데이터 쓰기/읽기 수행완료 여부에 상관 없이 send/recv 호출시 바로 함수를 반환한다. 실제 입출력이 이루어지는 동안 기다리는 것이 아니라 다른 일을 할 수 있게 된다.

 

비동기적 Notification

입출력 "알림"이 비동기적으로 이루어진다. select 함수의 비동기적 버전은 WSAEventSelect이며 입출력 완료 여부에 상관없이 함수를 바로 반환한다. 따라서 실제로 입출력 여부가 완료되었는지 확인하는 작업이 추가적으로 필요하다.

 

- WSAEventSelect 함수와 Notification

임의의 소켓을 대상으로 이벤트 발생여부의 관찰을 명령할 때 사용하는 함수

#include <winsock2.h>
int WSAEventSelect(SOCKET s, WSAEVENT hEventObject, long lNetworkEvents);

• s: 관찰대상인 소켓의 핸들 전달
• hEventObject: 이벤트 발생유무 확인을 위한 Event 오브젝트의 핸들 전달
• lNetworkEvents: 감시하고자 하는 이벤트의 유형 정보 전달

매개변수 s 소켓에 lNetworkEvents에 설정된 이벤트 정보를 등록하고 그 이벤트 중 하나가 발생하면 hEventObject에 전달된 핸들의 커널 오브젝트를 siganaled 상태로 바꾸는 함수이다. 리눅스에서 사용한 Epoll과 같은 효과를 얻으며 소켓에 이벤트 정보를 등록하여 이벤트가 발생하였을 때, hEventObject의 커널 오브젝트가 signal 상태가 된다.

 

lNetworkEvents에 전달가능한 인자

• FD_READ: 수신할 데이터가 있는가
• FD_WRITE: 블로킹 없이 데이터 전송이 가능한가
• FD_OOB: Out-of-band 데이터가 수신되었는가
• FD_ACCEPT 연결 요청이 있었는가
• FD_CLOSE: 연결의 종료가 요청되었는가 

 

- WSAEvent 오브젝트

이벤트 발생으로 WaitForMultipleObjects와 같은 함수로 signaled 상태가 확인된 후 non-signaled 상태가 자동적으로 되지 않는 manual-reset 모드의 이벤트 객체를 생성한다.

WSAEVENT WSACreateEvent(void);

• 성공 시 Event 오브젝트 핸들, 실패 시 WSA_INVALID_EVENT 반환

이벤트 객체를 소멸하는 방법은 다음과 같다.

BOOL WSACloseEvent(WSAEVENT hEvent);

• 성공 시 TRUE, 실패 시 FALSE 반환

- 이벤트 발생유무 확인

비동기적 모델에서 실제 이벤트가 발생하는 시점은 함수 호출 반환시점과 다르기 때문에 해당 소켓에 등록한 이벤트가 발생했는 지 확인하기 위해 함수를 호출해야한다.

 

DWORD WSAWaitForMultipleEvents(DWORD cEvents, const WSAEVENT* lphEvents, BOOL fWaitAll,
								DWORD dwTimeout, BOOL fAlertable);

• cEvents: signaled 상태로 전이여부를 호가인할 Event 오브젝트의 개수 정보 전달
• lphEvents: Event 오브젝트 핸들을 저장하고 있는 배열의 주소값 전달
• fWaitAll: TRUE 일 시 모든 Event가 Signal 상태일 때만 반환, FALSE시 하나만 signaled 상태가 되어도 반환
• dwTimeout, 1/1000초 단위로 타임아웃 지정. WSA_INFINITE 전달시 signaled 상태가 될때까지 반환하지 않는다.
• 반환 값: 반환된 정수값에서 상수 값 WSA_WAIT_EVENT_0를 빼면, 전달된 배열을 기준을 signaled 상태가 된 Event 오브젝트의 핸들이 저장된 인덱스가 계산된다. 둘 이상 이벤트가 발생되었으면 작은 인덱스가 반환되며 타임 아웃시 WAIT_TIMEOUT이 반환된다..

등록된 이벤트는 모두 manual-reset 모드이기 때문에 다음과 같이 이벤트가 발생한 오브젝트 핸들을 확인한다.

int posInfo = WSAWaitForMultipleEvents(numOfSock, hEventArray, FALSE, WSA_INFINITE, FALSE);
int startIdx = posInfo - WSA_WAIT_EVENT_0; // 첫번째 이벤트 발생한 id를 오프셋으로 빼줌
for(int i = startIdx; i < numOfSock; ++i)
{
	// 메뉴얼 리셋모드이기 때문에 여전히 signaled 상태이다.
	int sigEventIdx = WSAWaitForMultipleEvents(1, &hEventArray[i], TRUE, 0, FALSE);
    // ...
}

 

- 이벤트 종류의 구분

DWORD WSAEnumNetworkEvents(SOCKET s, WSAEVENT hEventObject, LPWSANETWORKEVENTS lpNetworkEvents);

• s: 이벤트가 발생한 소켓의 핸들 전달
• hEventObject: 소켓과 연결된 signaled 상태인 Event 오브젝트의 핸들 전달
• lpNetworkEvents: 발생한 이벤트의 유형 정보와 오류 정보로 채워질 WSANETWORKEVENTS 구조체 변수의 주소값 전달

위 함수 호출로 이벤트 객체의 signaled 상태를 non-signaled 상태로 되돌리니 이벤트 유형을 확인한 다음 ResetEvent 함수를 통해 이벤트 객체를 호출할 필요가 없다.

 

WSANETWORKEVENTS 구조체

typedef struct _WSANETWORKEVENTS
{
    long lNetworkEvents;
    int iErrorCode[FD_MAX_EVENTS];
} WSANETWORKEVENTS, *LPWSANETWORKEVENTS;

lNetworkEvents에는 발생한 이벤트 유형이 담기며, 에러가 발생했을 경우 iErrorCode에 0보다 큰 값이 저장된다.

 

따라서 발생한 이벤트 유형을 확인하고 에러가 발생했는 지 아니면 해당 이벤트를 처리하는 로직을 구성한다.

WSANETWORKEVENTS netEvents;
//...
WSAEnumNetworkEvents(hSock, hEvent, &netEvents);
if(netEvents.lNetworkEvents & FD_ACCEPT)
{
    if(netEvents.iErrorCode[FD_ACCEPT_BIT] != 0)
    {
    	// FD_ACCEPT 이벤트 관련 오류 발생
        break;
    }
	
    // 오류가 아니라면 연결 요청에 대한 이벤트 처리
}

 

비동기 Notification IO 모델의 서버 구현

 

1. 서버 소켓 생성 및 이벤트 등록

서버 소켓을 생성하고 WSAEvent 객체를 생성하여 클라이언트 요청에 대한 수락을 처리하기 위해 이벤트를 WSAEventSelect 함수를 통해 등록한다.

  // 전체 소켓 정보와 이벤트를 담을 배열
    SOCKET hSockArr[WSA_MAXIMUM_WAIT_EVENTS] = {};
    WSAEVENT hEventArr[WSA_MAXIMUM_WAIT_EVENTS] = {};

    // 서버 소켓 생성
    SOCKADDR_IN servAdr, clntAdr;
    SOCKET hServSock = socket(PF_INET, SOCK_STREAM, 0);
    memset(&servAdr, 0, sizeof(servAdr));
    servAdr.sin_family = AF_INET;
    servAdr.sin_addr.s_addr = htonl(INADDR_ANY);
    servAdr.sin_port = htons(PORT);

    if (bind(hServSock, (SOCKADDR*)&servAdr, sizeof(servAdr)) == SOCKET_ERROR)
        ErrorHandling("bind() error");

    if (listen(hServSock, 5) == SOCKET_ERROR)
        ErrorHandling("listen() error");

    // 서버 소켓에 non-signaled, manual reset 모드의 연결 요청 이벤트 등록
    WSAEVENT sockEvent = WSACreateEvent();
    if (WSAEventSelect(hServSock, sockEvent, FD_ACCEPT) == SOCKET_ERROR)
        ErrorHandling("WSAEventSelect() error");

    // 전체 소켓 정보와 이벤트 정보 저장
    int numOfClntSock = 0;
    hSockArr[numOfClntSock] = hServSock;
    hEventArr[numOfClntSock] = sockEvent;
    ++numOfClntSock;

2. 비동기적 IO를 확인하기 위해 WSAWaitForMultipleEvents를 호출한다.

만약 변화가 있다면 posInfo를 반환하고 startIdx를 계산하여 첫번째 signal 상태의 커널 오브젝트부터 처리한다. 타임 아웃과 실패 상태가 아니라면 발생한 이벤트 유형을 알아내기 위해 WSANETWORKEVENTS 객체를 생성해서 WSAEnumNetworkEvents 함수를 통해 상세 이벤트 내용을 추출한다.

    while (1)
    {
        int posInfo = WSAWaitForMultipleEvents(numOfClntSock, hEventArr, FALSE, WSA_INFINITE, FALSE);
        int startIdx = posInfo - WSA_WAIT_EVENT_0;

        for (int i = startIdx; i < numOfClntSock; ++i)
        {
            // 타임 아웃 혹은 실패 인지 먼저 체크
            int sigEventIdx = WSAWaitForMultipleEvents(1, &hEventArr[i], TRUE, 0, FALSE);
            if (sigEventIdx == WSA_WAIT_FAILED || sigEventIdx == WSA_WAIT_TIMEOUT)
            {
                continue;
            }
            else
            {
                WSANETWORKEVENTS netEvents = {};
                sigEventIdx = i; // 다시 원래 몇번 째 소켓인지 지정
                // 발생한 이벤트의 상세한 정보 요청
                WSAEnumNetworkEvents(hSockArr[sigEventIdx], hEventArr[sigEventIdx], &netEvents);

3.  이벤트 종류를 분리하고 처리한다.

클라이언트 연결 요청, 데이터 수신, 연결 종료 등 3가지의 유형에 따라 처리 로직을 구현하도록 한다.

                if (netEvents.lNetworkEvents & FD_ACCEPT) // 연결 요청 시
                {
                    if (netEvents.iErrorCode[FD_ACCEPT_BIT] != 0)
                    {
                        puts("Accept Error");
                        break;
                    }

                    // 연결 수락
                    int clntAdrLen = sizeof(clntAdr);
                    SOCKET hClntSock = accept(hSockArr[sigEventIdx], (SOCKADDR*)&clntAdr, &clntAdrLen);
                    sockEvent = WSACreateEvent();
                    WSAEventSelect(hClntSock, sockEvent, FD_READ | FD_CLOSE);

                    // 클라이언트 소켓 정보 저장
                    hEventArr[numOfClntSock] = sockEvent;
                    hSockArr[numOfClntSock] = hClntSock;
                    ++numOfClntSock;
                    puts("connected new client...");
                }

                if (netEvents.lNetworkEvents & FD_READ) // 데이터 수신시
                {
                    if (netEvents.iErrorCode[FD_READ_BIT] != 0)
                    {
                        puts("Read Error");
                        break;
                    }
                    int strLen = recv(hSockArr[sigEventIdx], msg, sizeof(msg), 0);
                    send(hSockArr[sigEventIdx], msg, strLen, 0);
                }


                if (netEvents.lNetworkEvents & FD_CLOSE) // 종료 요청 시
                {
                    if (netEvents.iErrorCode[FD_CLOSE_BIT] != 0)
                    {
                        puts("Close Error");
                        break;
                    }
                    WSACloseEvent(hEventArr[sigEventIdx]);
                    closesocket(hSockArr[sigEventIdx]);

                    --numOfClntSock;
                    CompressSockets(hSockArr, sigEventIdx, numOfClntSock);
                    CompressEvents(hEventArr, sigEventIdx, numOfClntSock);
                }