IOCP 서버 제작 실습 4

Send의 구조를 1-Send로 변경

• N-Send : 이전 비동기 Send에 대한 완료를 확인하지 않고 곧바로 다음 Send를 호출
• 1-Send : 이전 비동기 Send에 대한 완료가 확인되었을 때만 다음 Send를 호출
  따라서 Send할 데이터를 별도의 버퍼에 보관하는 과정이 필요함

 

• IOCPServer.h
  • stClientInfo 객체를 할당하여 관리하도록 mClientInfos를 vector<stClientInfo*>로 변경
  • StartServer() 함수 호출시 SendThread를 생성하도록 변경

//Send 쓰레드
std::thread					mSendThread;
//Send 쓰레드 동작 플래그
bool						mIsSenderRun = false;
    

void SendThread()
{
    while(mIsSenderRun)
    {
        for (auto client : mClientInfos)
        {
            if (client->IsConnectd()  == false)
            {
                continue;
            }

            client->SendIO();
        }

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

 

 

• ClientInfo.h
  • SendIO() 함수에서 실제로 WSASend() 함수를 통해 Send
    
    • mSendPos가 0이하일경우 버퍼에 보낼 데이터가 없는 것이므로 리턴
    • mIsSending 플래그가 켜져있는 경우 아직 이전 Send 작업 완료가 확인되지 않은 경우이므로 리턴
      Worker 쓰레드에서 SendCompleted()가 호출되어야 Send 작업 완료가 확인됨
    • 이전 Send 작업 완료가 확인되었고, 정상적으로 보낼 데이터가 존재하는 경우(mSendPos > 0)
      버퍼를 다른 쓰레드들과 공유하므로 lock_guard를 걸어주고, mIsSending 을 true로 바꾸어 Send 작업이 이루어지고있음을 표기
      이후 mSendBuf에 존재하는 데이터를 모두 mSendingBuf에 저장하고 WSASend() 함수를 호출하여 해당 데이터를 Send함
    • WSASend() 함수 호출 후에는 mSendPos를 0으로 초기화하여 Send 작업이 이루어졌음을 표기
  • SendMsg() 함수가 호출된경우 Recv한 데이터가 있는 것이므로 mSendBuf에 해당 데이터를 저장
    이 때도 마찬가지로 다른 쓰레드들과 공유하는 버퍼에 접근하는 것이므로 lock_guard를 걸어줌
    • 현재 mSendBuf에 존재하는 데이터 + 받은 데이터의 크기가 mSendBuf의 크기 이상인 경우 mSendPos를 0으로 변경하여 저장하는 링버퍼 형식이지만, 이 경우 한꺼번에 데이터가 많이 들어올 경우 데이터가 덮어씌워지는 문제가 생길 수 있음

bool SendIO()
{
    if (mSendPos <= 0 || mIsSending)
    {
        return true;
    }

    std::lock_guard<std::mutex> guard(mSendLock);

    mIsSending = true;

    CopyMemory(mSendingBuf, &mSendBuf[0], mSendPos);

    mSendOverlappedEx.m_wsaBuf.len = mSendPos;
    mSendOverlappedEx.m_wsaBuf.buf = &mSendingBuf[0];
    mSendOverlappedEx.m_eOperation = IOOperation::SEND;

    DWORD dwRecvNumBytes = 0;
    if (WSASend(mSock, &(mSendOverlappedEx.m_wsaBuf), 1,
        &dwRecvNumBytes, 0, (LPWSAOVERLAPPED) &(mSendOverlappedEx), NULL) == SOCKET_ERROR
        && (WSAGetLastError() != ERROR_IO_PENDING))
    {
        std::cout << "WSASend() Error : " << WSAGetLastError() << std::endl;
        return false;
    }

    mSendPos = 0;
    return true;
}

void SendCompleted(const UINT32 dataSize_)
{
    mIsSending = false;
    std::cout << "Send Bytes : " << dataSize_ << std::endl;
}

bool SendMsg(const UINT32 dataSize_, char *pMsg_)
{
    std::lock_guard<std::mutex> guard(mSendLock);

    if ((mSendPos + dataSize_) > MAX_SOCK_SENDBUF)
    {
        mSendPos = 0;
    }

    auto pSendBuf = &mSendBuf[mSendPos];

    CopyMemory(pSendBuf, pMsg_, dataSize_);
    mSendPos += dataSize_;

    return true;
}

 

 

• EchoServer.h
  • mPacketDataQueue에 쌓여있는 데이터가 있는 경우 SendMsg() 함수를 호출하여 최종적으로 각 클라이언트의 SendMsg() 함수를 호출

void ProcessPacket()
{
    while (mIsRunProcessThread)
    {
        auto packetData = DequePacketData();
        if (packetData.DataSize != 0)
        {
            SendMsg(packetData.SessionIndex, packetData.DataSize, packetData.pPacketData);
        }
        else
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
    }
}

 

 

---

---

 

 

1-Send를 일반 버퍼가 아닌 queue에서 작동하도록 변경

이전에 만들어두었던 ProcessPacket() 쓰레드에서 Send가 이루어지므로 이전에 별도로 만들었던 Send 쓰레드는 제거

 

• ClientInfo.h
  • SendMsg() 함수에서 mSendDataqueue에 Send할 데이터를 저장
    • mSendDataqueue는 공유 자원이므로 당연히 lock_guard를 걸어주어야함
    • queue의 크기가 1이라면 SendIO() 함수로 진입해 WSASend() 함수를 호출하여 실제로 Send
    • queue의 크기가 1이 아니라면 아직 이전 Send 작업 완료가 확인되지 않았으므로 아무 작업도 하지 않은 채 반환
  • SendCompleted() 함수가 호출된 경우 GetqueuedCompletionPort() 함수로 Send 작업 완료가 확인된 것이므로 mSendDataqueue의 원소를 지움
    • 역시 lock_guard를 걸어주어야함
    • mSendDataqueue에 들어있는 버퍼의 메모리와 해당 원소의 메모리 모두 해제가 필요
    • 메모리 해제 후 큐에서 원소를 제거
    • 제거 후에도 큐가 비어있지 않은 경우 추가로 Send를 해야하므로 SendIO() 함수를 호출하여 WSASend() 작업을 진행

std::mutex					mSendLock;
std::queue<stOverlappedEx*>	mSendDataqueue;

bool SendMsg(const UINT32 dataSize_, char *pMsg_)
{
    auto sendOverlappedEx = new stOverlappedEx;
    ZeroMemory(sendOverlappedEx, sizeof(stOverlappedEx));
    sendOverlappedEx->m_wsaBuf.len = dataSize_;
    sendOverlappedEx->m_wsaBuf.buf = new char[dataSize_];
    CopyMemory(sendOverlappedEx->m_wsaBuf.buf, pMsg_, dataSize_);
    sendOverlappedEx->m_eOperation = IOOperation::SEND;

    std::lock_guard<std::mutex> guard(mSendLock);

    mSendDataqueue.push(sendOverlappedEx);

    if (mSendDataqueue.size() == 1)
    {
        SendIO();
    }

    return true;
}

bool SendIO()
{
    auto sendOverlappedEx = mSendDataqueue.front();

    DWORD dwRecvNumBytes = 0;
    if (WSASend(mSock, &(sendOverlappedEx->m_wsaBuf), 1,
        &dwRecvNumBytes, 0, (LPWSAOVERLAPPED)sendOverlappedEx, NULL) == SOCKET_ERROR
        && (WSAGetLastError() != ERROR_IO_PENDING))
    {
        std::cout << "WSASend() Error : " << WSAGetLastError() << std::endl;
        return false;
    }

    return true;
}

void SendCompleted(const UINT32 dataSize_)
{
    std::cout << "Send Bytes : " << dataSize_ << std::endl;

    std::lock_guard<std::mutex> guard(mSendLock);

    delete[] mSendDataqueue.front()->m_wsaBuf.buf;
    delete mSendDataqueue.front();

    mSendDataqueue.pop();

    if (mSendDataqueue.empty() == false)
    {
        SendIO();
    }
}