IOCP 서버 제작 실습 1

https://www.youtube.com/watch?v=RMRsvll7hrM 

https://github.com/jacking75/edu_cpp_IOCP

GitHub - jacking75/edu_cpp_IOCP: IOCP 실습

위 링크의 IOCP 실습 자료를 통해 공부
단, IOCP 관련 지식은 따로 습득해야 링크의 단계별 실습을 따라갈 수 있음

 

 

기초적인 IOCP 에코 서버 구현

//main.cpp

#include "IOCompletionPort.h"

const UINT16 SERVER_PORT = 11021;
const UINT16 MAX_CLIENT = 100;

int main()
{
	IOCompletionPort ioCompletionPort;

	//소켓 초기화
	ioCompletionPort.InitSocket();

	//소켓 서버 등록
	ioCompletionPort.BindandListen(SERVER_PORT);

	ioCompletionPort.StartServer(MAX_CLIENT);

	std::cout << "아무 키나 누를 때까지 대기\n";
	getchar();

	ioCompletionPort.DestroyThread();
	return 0;
}

IOCompletionPort 클래스의 멤버변수들

class IOCompletionPort
{

private:
	// 클라이언트 정보 저장 구조체
	std::vector<stClientInfo>	mClientInfos;
	//리슨 소켓
	SOCKET						mListenSocket = INVALID_SOCKET;
	//접속중인 클라이언트 수
	int							mClientCnt = 0;
	//IO Worker 쓰레드
	std::vector<std::thread>	mIOWorkerThreads;
	//Accpet 쓰레드
	std::thread					mAccepterThread;
	//CompletionPort 객체 핸들
	HANDLE						mIOCPHandle = INVALID_HANDLE_VALUE;
	//작업 쓰레드 동작 플래그
	bool						mIsWorkerRun = true;
	//접속 쓰레드 동작 플래그
	bool						mIsAccepterRun = true;

...
}

 

Main 함수의 흐름을 따라가면서 각 함수를 설명

• InitSocket()
  • 가장 먼저 Winsock 사용을 위한 WSAStartup() 함수 호출
  • WSASocket() 함수로 서버 리슨소켓 생성

//소켓 초기화
bool InitSocket()
{
    WSADATA wsaData;

    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
    {
        std::cout << "WSAStartup() Error : " << WSAGetLastError() << std::endl;
        return false;
    }

    SOCKET mListenSocket = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, NULL, WSA_FLAG_OVERLAPPED);

    if (INVALID_SOCKET == mListenSocket)
    {
        std::cout << "WSASocket() Error : " << WSAGetLastError() << std::endl;
        return false;
    }

    std::cout << "socket init success\n";
    return true;
}

 

• BindandListen()
  • bind() 함수와 listen() 함수를 통한 서버 소켓 기본작업 수행

bool BindandListen(int nBindPort)
{
    SOCKADDR_IN			stServerAddr;
    stServerAddr.sin_family = AF_INET;
    stServerAddr.sin_port = htons(nBindPort);
    stServerAddr.sin_addr.s_addr = htonl(INADDR_ANY);

    if (bind(mListenSocket, (SOCKADDR*)&stServerAddr, sizeof(SOCKADDR_IN)) == SOCKET_ERROR)
    {
        std::cout << "bind() Error : " << WSAGetLastError() << std::endl;
        return false;
    }

    if (listen(mListenSocket, 5) == SOCKET_ERROR)
    {
        std::cout << "listen() Error : " << WSAGetLastError() << std::endl;
        return false;
    }

    std::cout << "server registration success\n";
    return true;
}

 

• StartServer()
  • CreateClient() 함수로 클라이언트 생성
  • CreateIoCompletionPort() 함수로 IOCP 오브젝트 핸들 생성
  • CreateWorkerThread()로 I/O 작업을 처리할 작업자 쓰레드 생성
  • CreateAccepterThread()로 사용자 접속을 처리할 쓰레드 생성

//접속 요청 수락, 메시지 처리
bool StartServer(const UINT32 maxClientCount)
{
    CreateClient(maxClientCount);

    mIOCPHandle = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, MAX_WORKERTHREAD);
    if (mIOCPHandle == NULL)
    {
        std::cout << "CreateIoCompletionPort() Error : " << GetLastError() << std::endl;
        return false;
    }

    if (CreateWorkerThread() == false)
    {
        return false;
    }

    if (CreateAccepterThread() == false)
    {
        return false;
    }

    std::cout << "Server start\n";
    return true;
}

 

• CreateClient()
  • emplace_back()
    C++11에서 추가된 멤버함수, push_back()과 마찬가지로 vector의 요소 끝에 원소를 추가
    단, 객체를 받는 것이 아닌 삽입할 객체의 생성자에 사용되는 인자를 받아 vector 내에서 직접 객체를 생성
    따라서 임시 객체 생성 및 파괴 과정을 거치지 않아 성능상 유리함
  • ZeroMemory
    Window에서 사용되는 매크로, 메모리 영역이 0으로 채워지는것이 명시되어있는 점을 제외하고는 memset()과 동일하게 동작(내부적으로 memset() 호출)

void CreateClient(const UINT32 maxClientCount)
{
    for (UINT32 i = 0; i < maxClientCount; ++i)
    {
        mClientInfos.emplace_back();
    }
}

//클라이언트 정보 구조체
struct stClientInfo
{
	SOCKET			m_socketClient;			//Client 소켓
	stOverlappedEx	m_stRecvOverlappedEx;	//Recv Overlapped I/O 작업 위한 변수
	stOverlappedEx	m_stSendOverlappedEx;	//Send OVerlapped I/O 작업 위한 변수

	stClientInfo()
	{
		ZeroMemory(&m_stRecvOverlappedEx, sizeof(stOverlappedEx));
		ZeroMemory(&m_stSendOverlappedEx, sizeof(stOverlappedEx));
		m_socketClient = INVALID_SOCKET;
	}
};

//WSAOVERLAPPED 구조체 확장
struct stOverlappedEx
{
	WSAOVERLAPPED	m_wsaOverlapped;		//OVERLAPPED 구조체
	SOCKET			m_socketClient;			//클라이언트 소켓
	WSABUF			m_wsaBuf;				//Overlapped I/O 작업 버퍼
	char			m_szBuf[MAX_SOCKBUF];	//데이터 버퍼
	IOOperation		m_eOperation;			//작업 동작 종류
};

 

 

• CreateWorkerThread()
  • 쓰레드 생성에 lambda 표현식을 사용, [this]로 람다식이 속하는 클래스 오브젝트를 참조로 캡쳐
• WorkerThread() 
  • GetQueuedCompletionStatus() 함수로 Completion Port(예제에서는 mIOCPHandle 변수로 지정)에 지정되어있는 소켓을 감시
    • AccepterThread()에서 클라이언트 소켓을 Completion Port에 등록할 때 사용한 ConpletionKey를 pClientInfo로 받아와서 사용
    • 비동기 입출력 함수 호출시 사용한 OVERLAPPED 구조체를 lpOverlapped로 받아옴
    • 마지막 인자가 INFINITE이므로 입출력 완료 패킷이 생성되어 운영체제가 쓰레드를 깨울 때까지 무한정 대기함
  • GetQueuedCompletionStatus() 함수 결과에 따른 케이스 처리 (추후 보완 필요)
    • 'bSuccess == TRUE && dwIoSize == 0 && lpOverlapped == NULL' 인 경우
      I/O 작업이 완료된 상태이므로 쓰레드 종료
    • 'bSuccess == FALSE && lpOverlapped == NULL'
      completion port에 완료된 I/O 작업이 없으므로 다시 GetQueuedCompletionStatus()로 돌아가 대기
    • 'bSuccess == FALSE && lpOverlapped != NULL' 인 경우
      GetQueuedConpletionStatus() 함수에 에러가 발생한 상황이므로 클라이언트 접속 해제
      GetLastError() 함수로 에러 사항을 확인할 수도 있음
    • 'bSuccess == TRUE && dwIoSize == 0 && lpOverlapped != NULL' 인 경우
      zero-byte operation, 클라이언트 접속 해제
  • 처리한 비동기 I/O 작업의 종류에 따른 처리 
    • Recv라면 수신한 정보 출력 후 SendMsg() 함수로 에코, BindRecv() 함수로 다시 Recv
    • Send라면 송신한 정보만 출력 
• CloseSocket()
  • 오류가 발생했을 때 강제 종료
  • struct linger : TCP 소켓 옵션인 SO_LINGER에 사용되는 구조체 
    l_onoff - Linger 옵션 온오프 플래그 / l_linger - 옵션 활성화시 기다리는 시간
    
    • l_onoff == 0
      소켓의 디폴트값, 소켓 버퍼에 남아있는 모든 데이터를 전송하는 일반적인 소켓의 정상 종료
    • l_onoff > 0, l_linger == 0
      close() 함수 호출시 상대방에게 RST 패킷을 전송, 소켓 버퍼에 남아있는 데이터를 버리는 비정상 종료가 이루어짐
    • l_onoff > 0, l_linger > 0
      close() 함수 호출 후 송신 버퍼에 남아있는 데이터와 FIN 패킷을 전송
      이후 지정된 시간동안 대기, 시간 안에 FIN_ACK 패킷을 받지 못하면 EWOLULDBLOCK 반

//worker 쓰레드 생성
bool CreateWorkerThread()
{
    for (int i = 0; i < MAX_WORKERTHREAD; i++)
    {
        mIOWorkerThreads.emplace_back([this]() {WorkerThread(); });
    }

    std::cout << "WorkerThread start...\n";
    return true;
}

//worker Thread
void WorkerThread()
{
    stClientInfo*	pClientInfo = NULL;
    BOOL			bSuccess = TRUE;
    DWORD			dwIoSize = 0;
    LPOVERLAPPED	lpOverlapped = NULL;

    while (mIsWorkerRun)
    {
        bSuccess = GetQueuedCompletionStatus(mIOCPHandle, &dwIoSize, (PULONG_PTR)&pClientInfo, &lpOverlapped, INFINITE);

        //쓰레드 종료 메시지 처리
        if (bSuccess == TRUE && dwIoSize == 0 && lpOverlapped == NULL)
        {
            mIsWorkerRun = false;
            continue;
        }

        if (lpOverlapped == NULL)
        {
            continue;
        }

        // 클라이언트 접속 해제
        if (bSuccess == FALSE || (dwIoSize == 0 && bSuccess == TRUE))
        {
            std::cout << "socket " << (int)pClientInfo->m_socketClient << " disconnected\n";
            CloseSocket(pClientInfo);
            continue;
        }

        stOverlappedEx* pOverlappedEx = (stOverlappedEx*)lpOverlapped;

        //Overlapped I/O Recv 
        if (pOverlappedEx->m_eOperation == IOOperation::RECV)
        {
            pOverlappedEx->m_szBuf[dwIoSize] = NULL;
            std::cout << "Recv Bytes : " << dwIoSize << ", msg : " << pOverlappedEx->m_szBuf << std::endl;

            //echo
            SendMsg(pClientInfo, pOverlappedEx->m_szBuf, dwIoSize);
            BindRecv(pClientInfo);
        }
        //Overlapped I/O Send
        else if (pOverlappedEx->m_eOperation == IOOperation::SEND)
        {
            std::cout << "Send Bytes : " << dwIoSize << ", msg : " << pOverlappedEx->m_szBuf << std::endl;
        }
        else
        {
            std::cout << "socket " << (int)pClientInfo->m_socketClient << " exception";
        }
    }
}

//소켓 연결 종료
void CloseSocket(stClientInfo* pClientInfo, bool bIsForce = false)
{
    // SO_DONTLINGER로 설정
    struct linger stLinger = { 0, };

    // SO_LINGER, timeout = 0으로 설정하여 강제 종료, 데이터 손실이 있을 수 있음
    if (bIsForce == true)
    {
        stLinger.l_onoff = 1;
    }

    shutdown(pClientInfo->m_socketClient, SD_BOTH);

    setsockopt(pClientInfo->m_socketClient, SOL_SOCKET, SO_LINGER, (char*)&stLinger, sizeof(stLinger));

    closesocket(pClientInfo->m_socketClient);

    pClientInfo->m_socketClient = INVALID_SOCKET;
}

 

• CreateAccepterThread()
  • 쓰레드 생성에 lambda 표현식을 사용, [this]로 람다식이 속하는 클래스 오브젝트를 참조로 캡쳐
•  AccepterThread()
  • accept() 함수에서 서버로의 연결 요청이 들어올 때까지 대기
  • accept 성공 후 BindIOCompletionPort() 함수에서 Completion Port에 클라이언트 등록
  • BindRecv() 함수에서 Recv 요청

//accept요청 처리 쓰레드 생성
bool CreateAccepterThread()
{
    mAccepterThread = std::thread([this]() {AccepterThread(); });

    std::cout << "AccepterThread start...\n";
    return true;
}

//사용자 접속 쓰레드
void AccepterThread()
{
    SOCKADDR_IN		stClientAddr;
    int				nAddrLen = sizeof(SOCKADDR_IN);

    while (mIsAccepterRun)
    {
        stClientInfo* pClientInfo = GetEmptyClientInfo();
        if (pClientInfo == NULL)
        {
            std::cout << "Client Full Error\n";
            return;
        }

        pClientInfo->m_socketClient = accept(mListenSocket, (SOCKADDR*)&stClientAddr, &nAddrLen);
        if (pClientInfo->m_socketClient == INVALID_SOCKET)
        {
            continue;
        }

        //I/O Completion Port 객체와 소켓 연결
        if (BindIOCOmpletionPort(pClientInfo) == false)
        {
            return;
        }

        //Recv Overlapped I/O 작업 요청
        if (BindRecv(pClientInfo) == false)
        {
            return;
        }

        char clientIP[32] = { 0, };
        inet_ntop(AF_INET, &(stClientAddr.sin_addr), clientIP, 32 - 1);
        std::cout << "Client connect : IP(" << clientIP << ") SOCKET(" << (int)pClientInfo->m_socketClient << ")\n";

        ++mClientCnt;
    }
}

//CompletionPort 객체와 소켓 연결
bool BindIOCompletionPort(stClientInfo* pClientInfo)
{
    auto hIOCP = CreateIoCompletionPort((HANDLE)pClientInfo->m_socketClient, mIOCPHandle, (ULONG_PTR)(pClientInfo), 0);

    if (hIOCP == NULL || hIOCP != mIOCPHandle)
    {
        std::cout << "CreateIoCompletionPort() Error : " << GetLastError() << std::endl;
        return false;
    }

    return true;
}

//WSARecv overlapped I/O
bool BindRecv(stClientInfo* pClientInfo)
{
    DWORD dwFlag = 0;
    DWORD dwRecvNumBytes = 0;

    pClientInfo->m_stRecvOverlappedEx.m_wsaBuf.len = MAX_SOCKBUF;
    pClientInfo->m_stRecvOverlappedEx.m_wsaBuf.buf = pClientInfo->m_stRecvOverlappedEx.m_szBuf;
    pClientInfo->m_stRecvOverlappedEx.m_eOperation = IOOperation::RECV;

    if (WSARecv(pClientInfo->m_socketClient, &(pClientInfo->m_stRecvOverlappedEx.m_wsaBuf), 1,
        &dwRecvNumBytes, &dwFlag, (LPWSAOVERLAPPED) & (pClientInfo->m_stRecvOverlappedEx), NULL) == SOCKET_ERROR
        && (WSAGetLastError() != ERROR_IO_PENDING))
    {
        std::cout << "WSARecv() Error : " << WSAGetLastError() << std::endl;
        return false;
    }

    return true;
}

 

• DestroyThread()
  • worker 쓰레드와 accepter 쓰레드들을 돌면서 join()

//생성되어있던 쓰레드 파괴
void DestroyThread()
{
    mIsWorkerRun = false;
    CloseHandle(mIOCPHandle);

    for (auto& th : mIOWorkerThreads)
    {
        if (th.joinable())
        {
            th.join();
        }
    }

    mIsAccepterRun = false;
    closesocket(mListenSocket);

    if (mAccepterThread.joinable())
    {
        mAccepterThread.join();
    }
}