IOCP 서버 제작 실습 5

Accept 처리를 비동기 I/O로 변경

 

• IOCPServer.h
  • 리슨소켓의 비동기 I/O 작업 완료 여부를 감시해야하므로 BindandListen() 함수에서 CreateIoCompletionPort() 함수를 통해 mListenSocket을 Completion Port에 등록
  • AccepterThread()에서 클라이언트들을 돌면서 비동기 I/O Accept 작업을 수행
    • 이미 클라이언트가 연결되어있다면(IsConnected() == true) 연결 시도를 하지 않고 넘어감
    • std::chrono
      • std::chrono::steady_clock은 프로세스마다 독립적으로 측정되는 시간 측정 방법을 사용하여 다른 프로세스나 시스템의 시간 변화에 영향을 받지 않음.
      • 1970년 1월 1일을 기준으로 하는 system_clock에서의 epoch(기준 시각)과는 달리 steady_clock에서의 epoch는 라이브러리 구현체마다 다를 수 있기 때문에 상대적인 시간 간격에서만 활용해야함
    • curTimeSec이 client->GetLatestClosedTimeSec()보다 작을 경우 아직 클라이언트의 연결이 해제된 상황이 아니기 때문에 연결 시도를 하지 않고 넘어감
    • curTimeSec과 client->GetLatestClosedTimeSec() 차가 양수이지만 RE_USE_SESSION_WAIT_TIMESEC 보다 작을 경우 클라이언트의 연결 해제 이후 시간이 얼마 지나지 않은 것이므로 재연결 시도를 하지 않고 넘어감
    • 그렇지 않다면 PostAccept() 함수를 호출하여 비동기 I/O Accept 작업을 수행
  • Worker 쓰레드에서 GetqueuedCompletionStatus() 함수로 m_eOperation이 ACCEPT인 경우를 찾아내면 비동기 I/O Accept 작업이 완료된 것이므로 ACceptCompletion() 함수로 넘어가 이후 필요한 작업들을 진행

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;
    }

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


    if (CreateIoCompletionPort((HANDLE)mListenSocket, mIOCPHandle, (UINT32)0, 0) == nullptr)
    {
        std::cout << "listen socket IOCP bind error : " << WSAGetLastError() << std::endl;
        return false;
    }

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

//사용자 접속 쓰레드
void AccepterThread()
{
    while (mIsAccepterRun)
    {
        auto curTimeSec = std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now().time_since_epoch()).count();

        for (auto client : mClientInfos)
        {
            if (client->IsConnectd())
            {
                continue;
            }

            auto diff = (UINT64)curTimeSec - client->GetLatestClosedTimeSec();
            if (diff < 0)
            {
                continue;
            }
            else if (diff <= RE_USE_SESSION_WAIT_TIMESEC)
            {
                continue;
            }

            client->PostAccept(mListenSocket, curTimeSec);

        }

        std::this_thread::sleep_for(std::chrono::milliseconds(32));
    }
}

//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;
        }

        auto pOverlappedEx = (stOverlappedEx*)lpOverlapped;

        // 클라이언트 접속 해제
        if (bSuccess == FALSE || (dwIoSize == 0 && pOverlappedEx->m_eOperation != IOOperation::ACCEPT))
        {
            CloseSocket(pClientInfo);
            continue;
        }

        // 클라이언트 등록
        if (pOverlappedEx->m_eOperation == IOOperation::ACCEPT)
        {
            pClientInfo = GetClientInfo(pOverlappedEx->SessionIndex);
            if (pClientInfo->AcceptCompletion())
            {
                ++mClientCnt;

                OnConnect(pClientInfo->GetIndex());
            }
            else
            {
                CloseSocket(pClientInfo, true);
            }
        }
        //Overlapped I/O Recv 
        else if (pOverlappedEx->m_eOperation == IOOperation::RECV)
        {
            OnReceive(pClientInfo->GetIndex(), dwIoSize, pClientInfo->GetRecvBuffer());

            pClientInfo->BindRecv();
        }
        //Overlapped I/O Send
        else if (pOverlappedEx->m_eOperation == IOOperation::SEND)
        {
            pClientInfo->SendCompleted(dwIoSize);
        }
        else
        {
            std::cout << "socket " << (int)pClientInfo->GetIndex() << " exception";
        }
    } 
}

 

 

• ClientInfo.h
  • PostAccept() 함수에서는 WSASocket() 함수로 클라이언트 소켓을 생성한 후 AcceptEx() 함수로 비동기 I/O Accept 작업을 진행
  • AcceptCompletion() 함수는 Accept 작업이 완료된 이후 호출되므로 클라이언트 소켓을 Completion Port에 등록하고 Recv를 요청
  • 클라이언트가 연결 해제될 때는 mLatestClosedTimeSec의 값을 갱신하여 클라이언트의 연결이 해제된 시점을 알 수 있도록 함

bool PostAccept(SOCKET listenSock_, const UINT64 curTimeSec_)
{
    std::cout << "PostAccept. client Index : " << GetIndex() << std::endl;

    mLatestClosedTimeSec = UINT32_MAX;

    mSock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_IP, NULL, 0, WSA_FLAG_OVERLAPPED);
    if (mSock == INVALID_SOCKET)
    {
        std::cout << "Client WSASocket Error : " << GetLastError() << std::endl;
        return false;
    }

    ZeroMemory(&mAcceptContext, sizeof(stOverlappedEx));

    DWORD bytes = 0;
    DWORD flags = 0;
    mAcceptContext.m_wsaBuf.len = 0;
    mAcceptContext.m_wsaBuf.buf = nullptr;
    mAcceptContext.m_eOperation = IOOperation::ACCEPT;
    mAcceptContext.SessionIndex = mIndex;

    if (AcceptEx(listenSock_, mSock, mAcceptBuf, 0, sizeof(SOCKADDR_IN) + 16,
        sizeof(SOCKADDR_IN) + 16, &bytes, (LPWSAOVERLAPPED) & (mAcceptContext)) == FALSE
        && (WSAGetLastError() != WSA_IO_PENDING))
    {
        std::cout << "AcceptEx Error : " << GetLastError() << std::endl;
        return false;
    }

    return true;
}

bool AcceptCompletion()
{
    std::cout << "AcceptCompletion : SessionIndex " << mIndex << std::endl;

    if (OnConnect(mIOCPHandle, mSock) == false)
    {
        return false;
    }

    SOCKADDR_IN stClientAddr;
    int nAddrLen = sizeof(SOCKADDR_IN);
    char clientIP[32] = { 0, };
    inet_ntop(AF_INET, &(stClientAddr.sin_addr), clientIP, 32 - 1);
    std::cout << "Client Connected : IP " << clientIP << " SOCKET " << (int)mSock << std::endl;

    return true;
}

bool OnConnect(HANDLE iocpHandle_, SOCKET socket_)
{
    mSock = socket_;
    mIsConnect = 1;

    Clear();

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

    //Recv Overlapped I/O 작업 요청
    return BindRecv();
}

void Close(bool bIsForce = false)
{
    // SO_DONTLINGER로 설정
    struct linger stLinger = { 0, 0 };

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

    shutdown(mSock, SD_BOTH);

    setsockopt(mSock, SOL_SOCKET, SO_LINGER, (char*)&stLinger, sizeof(stLinger));

    mIsConnect = 0;
    mLatestClosedTimeSec = std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now().time_since_epoch()).count();

    closesocket(mSock);
    mSock = INVALID_SOCKET;
}