TCP와 링버퍼

TCP와 L7 버퍼링

TCP와 UDP를 비교하였을 때, 개발 측면에서의 가장 큰 차이점은 UDP는 데이터그램 방식으로서 송신한 메시지가 정확한 크기로 상대방에게 전달되지만, TCP는 데이터 간 경계가 없이 스트림 방식으로 송수신된다는 점입니다.

이러한 특징으로 인하여 7계층에서 4계층 TCP 버퍼로부터 데이터를 읽어올 때, 메시지의 완전한 형태의 패킷으로 받아지지 않는 경우를 대비하기 위해 7계층에도 수신 버퍼가 필요합니다.

7계층에서 수신된 모든 네트워크 데이터를 버퍼링하고, 해당 데이터가 충분히 쌓이게 되어 읽을 수 있는 상태가 된다면 그 때 메시지를 읽어내어 버퍼에서 빼내는 방식을 사용해야 합니다.

L7 버퍼의 필요성

수신 링버퍼는 TCP의 특성상 데이터의 경계가 없이 스트림으로 오는 수신 데이터를 처리하기 위해 필요하다고 논의했습니다. 이에 더불어, L7 계층에 추가적인 버퍼를 둠으로써 얻을 수 있는 이점이 더 있습니다. L4계층의 소켓 당 버퍼는 유한합니다. L4 계층에서 TCP는 송신측에 자신의 버퍼 여유량을 알려줌으로서 (윈도우 사이즈), 전송량을 효율적으로 제어합니다. 따라서, L4계층의 소켓 수신 버퍼가 다 찼다면, 상대방도 데이터 전송이 불가능해집니다.

따라서 응용어플리케이션은 소켓의 수신버퍼에서 데이터를 최대한 빠르게 비워내주어야 합니다. 그래야만 윈도우 사이즈의 감소로 인한 송수신 속도 감소를 방지할 수 있고, 이중 버퍼링으로 더욱 많은 양의 데이터를 유연하게 처리해 줄 수 있습니다.
송신 링버퍼 또한 L4 계층의 소켓 송신 버퍼가 꽉 찼을 경우에 한번 더 버퍼링 해주는 역할로서 가치가 있습니다. 물론, L4 계층의 소켓 송신 버퍼가 다 차기 위해서는, 상대방의 수신 버퍼가 모두 차고, 이후 내 L4 소켓 송신 버퍼가 모두 차야 하는 상황이 될 것이라 흔하지는 않습니다.

마지막으로, 송수신 버퍼를 두는 것은 send()와 recv() 호출을 줄이는 가장 편리한 방법이 될 수 있습니다. 전송이 필요할 때마다 send()를 호출하고, 수신이 필요할 때마다 recv()를 호출하는 것은 커널 모드로의 전환 때문에 성능상 비용이 클 뿐만 아니라, 예외 처리 관리 또한 힘들어집니다. 따라서 성능적으로도, 구조적으로도 L7 버퍼는 필수불가결입니다.

링버퍼

링버퍼의 구조

링버퍼는 원형 큐와 유사하게 맨 앞 버퍼와 맨 뒤 버퍼가 연결될 수 있도록 한 원형 버퍼입니다.
네트워크 메시지 처리에 사용하는 버퍼이기에, 바이트 단위로 넣고 뺄 수 있는 인터페이스 또한 필수로 가지고 있어야 합니다.

기본 인터페이스 설계

// 링버퍼의 총 버퍼 크기를 리턴합니다
unsigned int GetSizeTotal() const;
// 링버퍼에서 남은 버퍼 크기를 리턴합니다
unsigned int GetSizeFree() const;
// 링버퍼에서 사용중인 버퍼 크기를 리턴합니다
unsigned int GetSizeUsed() const;

// 링버퍼에 데이터를 인큐합니다
bool Enqueue(const char *data, unsigned int requestSize, unsigned int *outEnqueueSize, bool isPartialEnqueueAvailable = false);
// 링버퍼로부터 데이터를 디큐합니다
bool Dequeue(char *outData, unsigned int requestSize, unsigned int *outDequeueSize, bool isPartialDequeueAvailable = true, bool isPeekMode = false);
// 링버퍼로부터 데이터를 디큐하지 않고 읽어옵니다
bool Peek(char *outData, unsigned int requestSize, unsigned int *outPeekSize, bool isPartialPeekAvailable = true);

bool IsEmpty() const;
bool IsFull() const;
void ClearBuffer();

링버퍼의 사용 (기본)

// 세션별로 송수신 링버퍼를 생성합니다
player->recvBuffer = new RingBuffer();
player->sendBuffer = new RingBuffer();

// L4계층의 TCP 데이터를 L7 링버퍼로 수신합니다.
char buffer[BUFFER_MAX_SIZE];
int recvResult = recv(player->socket, buffer, BUFFER_MAX_SIZE, 0);
if (recvResult > 0) 
{
    unsigned int enqueueResultSize = 0;
    player->recvBuffer.Enqueue(buffer, recvResult, &enqueueResultSize);
}

// L7 링버퍼에서 메시지 헤더 크기만큼 받아오기를 시도합니다
BOOL peekResult = player->recvBuffer->Peek(reinterpret_cast<PCHAR>(&netHeader), sizeof(NetHeader), &peekOutSize, false);

링버퍼 사용 성능 개선

위 사용 샘플 코드에서, 소켓으로부터 데이터 수신 시 임시 버퍼를 거쳐서 링버퍼로 들어감으로서, 복사가 두 번 이루어지는 비효율적인 동작이 일어나게 됩니다.
recv() 함수에 임시 버퍼가 아닌, 링버퍼의 실제 버퍼 포인터를 넣는 식으로 개선해볼 수 있지만, 이 경우 몇 가지 기능을 추가적으로 제공하도록 개선해야 합니다.

우선, 링버퍼의 실제 버퍼 포인터를 외부로 제공해야 합니다.

// 링버퍼의 읽기 포인터를 리턴합니다
char *GetFrontBuffer() const;
// 링버퍼의 쓰기 포인터를 리턴합니다
char *GetRearBuffer() const;

외부로 포인터 자체를 제공하는 것이기 때문에, 네트워크 라이브러리 엔진 내부에서만 사용 가능하도록 private 한정자에 friend 처리를 해 주어 관리하는 것이 바람직합니다.
이를 통해 recv() 나 send() 함수에 곧바로 링버퍼의 버퍼 포인터를 전달할 수 있지만, 해당 버퍼 포인터 위치부터 메모리 연속적으로 쓰거나 읽기 가능한 크기는 링버퍼의 끊긴 위치까지입니다.

// 링버퍼에서 끊기지 않고 연속으로 접근 가능하게끔 인큐 가능한 크기를 리턴합니다
unsigned int GetSizeDirectEnqueueAble() const;
// 링버퍼에서 끊기지 않고 연속으로 접근 가능하게끔 디큐 가능한 크기를 리턴합니다
unsigned int GetSizeDirectDequeueAble() const

따라서 위 기능을 인터페이스를 통해 제공해야 링버퍼의 버퍼 포인터를 사용할 수 있습니다.

또한, Enqueue() 나 Dequeue()를 통해 정상적으로 버퍼 포인터가 다루어졌다면 그에 맞추어 링버퍼의 front와 rear 포지션이 재계산됩니다. 하지만 이 경우 버퍼 포인터를 통해 직접적으로 메모리 접근이 이루어져 링버퍼의 front 와 rear 포지션이 변화가 없이 옛 상태 기준이므로, 이를 적절히 수정해주어야 합니다.

// 링버퍼의 읽기 포인터를 뒤쪽으로 이동시킵니다
bool MoveFrontBufferRear(unsigned int moveSize);
// 링버퍼의 쓰기 포인터를 뒤쪽으로 이동시킵니다
bool MoveRearBufferRear(unsigned int moveSize);

위 세 가지 기능을 제공하는 링버퍼는, 송수신에 임시 버퍼를 필요로 하지 않고, 직접적으로 recv()와 send()에 버퍼 포인터를 통해 쓰고 읽을 수 있게 성능을 개선시킬 수 있습니다.

링버퍼의 사용 (개선후)

// 링버퍼에서 스트림으로 읽어올 수 있는 만큼 4계층에서 데이터를 읽어옵니다.
int recvResult = recv(player->socket, player->recvBuffer->GetRearBuffer(), player->recvBuffer->GetSizeDirectEnqueueAble(), 0);

// 링버퍼에서 스트림으로 써나갈 수 있는 만큼 4계층으로 데이터를 전송합니다.
int sendResult = send(player->socket, player->sendBuffer->GetFrontBuffer(), player->sendBuffer->GetSizeDirectDequeueAble(), 0);

코드 모음

링버퍼에 데이터를 인큐하는 코드

/**
 * \brief Insert Buffer To Queue
 * \param data inserting buffer
 * \param requestSize inserting buffer size
 * \param outEnqueueSize [out] successfully inserted size
 * \param isPartialEnqueueAvailable if true, partial data insertion will be allowed
 * \return is buffer insertion success
 */
bool RingBuffer::Enqueue(const char* data, unsigned int requestSize, unsigned int * outEnqueueSize, bool isPartialEnqueueAvailable)
{
    unsigned int addedSize = 0;
    unsigned int handleSize = std::min(requestSize, GetSizeFree());

    // HANDLE PARAMETER EXCEPTIONS
    if (IsFull() || requestSize <= 0 || data == nullptr)
    {
        if (outEnqueueSize != nullptr) memcpy(outEnqueueSize, &addedSize, sizeof(unsigned int));
        return false;
    }

    // IS PARTIAL RESTRICTED MODE, CHECK INSERTION IS AVAILABLE
    if (!isPartialEnqueueAvailable && handleSize != requestSize)
    {
        if (outEnqueueSize != nullptr) memcpy(outEnqueueSize, &addedSize, sizeof(unsigned int));
        return false;
    }

    // CHECK DATA WILL BE SPLIT
    if (handleSize > GetSizeDirectEnqueueAble())
    {
        // DATA WILL BE SPLIT
        const unsigned int splitLeftSize = GetSizeDirectEnqueueAble();
        const unsigned int splitRightSize = handleSize - splitLeftSize;
        const char *dataLeft = data;
        const char *dataRight = data + splitLeftSize;

        // INSERT PARTIAL LEFT
        memcpy(GetRearBuffer(), dataLeft, splitLeftSize);
        rearPosition_ += splitLeftSize;
        if (rearPosition_ > GetEndBuffer()) {
            rearPosition_ -= (bufferSize_ + 1);
        }
        addedSize += splitLeftSize;

        // INSERT PARTIAL RIGHT
        memcpy(GetRearBuffer(), dataRight, splitRightSize);
        rearPosition_ += splitRightSize;
        addedSize += splitRightSize;
    } else
    {
        // DATA WILL NOT SPLIT
        memcpy(GetRearBuffer(), data, handleSize);
        rearPosition_ += handleSize;
        if (rearPosition_ > GetEndBuffer()) {
            rearPosition_ -= (bufferSize_ + 1);
        }
        addedSize += handleSize;
    }

    usedSize_ += addedSize;
    if (outEnqueueSize != nullptr) memcpy(outEnqueueSize, &addedSize, sizeof(unsigned int));
    return true;
}

링버퍼에 데이터를 디큐하는 코드

/**
 * \brief Retrieve Buffer From Queue
 * \param outData [out] data retrieve dest
 * \param requestSize requesting buffer size
 * \param outDequeueSize [out] successfully retrieved size
 * \param isPartialDequeueAvailable if true, partial data retrieval will be allowed
 * \param isPeekMode if true, just copy to dest and do not remove from queue
 * \return is buffer retrieval success
 */
bool RingBuffer::Dequeue(char* outData, unsigned int requestSize, unsigned int * outDequeueSize, bool isPartialDequeueAvailable, bool isPeekMode)
{
    unsigned int removedSize = 0;
    unsigned int handleSize = std::min(requestSize, GetSizeUsed());

    // HANDLE PARAMETER EXCEPTIONS
    if (IsEmpty() || requestSize <= 0 || outData == nullptr)
    {
        if (outDequeueSize != nullptr) memcpy(outDequeueSize, &removedSize, sizeof(unsigned int));
        return false;
    }

    // IS PARTIAL RESTRICTED MODE, CHECK REMOVAL IS AVAILABLE
    if (!isPartialDequeueAvailable && handleSize != requestSize)
    {
        if (outDequeueSize != nullptr) memcpy(outDequeueSize, &removedSize, sizeof(unsigned int));
        return false;
    }

    // CHECK DATA WILL BE SPLIT
    if (handleSize > GetSizeDirectDequeueAble())
    {
        // DATA RETRIEVAL WILL BE SPLIT
        unsigned int retrievalSize = 0;
        char *frontTempBuffer = GetFrontBuffer();

        // GET DATA FRONT ~ END
        retrievalSize += static_cast<unsigned int>(GetEndBuffer() - frontTempBuffer) + 1;
        memcpy(outData, frontTempBuffer, retrievalSize);
        removedSize += retrievalSize;
        frontTempBuffer += removedSize;
        if (frontTempBuffer > GetEndBuffer()) {
            frontTempBuffer -= (bufferSize_ + 1);
        }
        if (!isPeekMode) {
            frontPosition_ += retrievalSize;
            if (frontPosition_ > GetEndBuffer()) {
                frontPosition_ -= (bufferSize_ + 1);
            }
        }
        // GET DATA START ~ REQUEST SIZE
        retrievalSize = handleSize - removedSize;
        memcpy(outData + removedSize, frontTempBuffer, retrievalSize);
        removedSize += retrievalSize;
        if (!isPeekMode)
        {
            frontPosition_ += retrievalSize;
        }
    } else
    {
        // DATA RETRIEVAL IS NOT SPLIT
        memcpy(outData, GetFrontBuffer(), handleSize);
        removedSize += handleSize;
        if (!isPeekMode)
        {
            frontPosition_ += handleSize;
            if (frontPosition_ > GetEndBuffer()) {
                frontPosition_ -= (bufferSize_ + 1);
            }
        }
    }

    if (!isPeekMode) usedSize_ -= removedSize;
    if (outDequeueSize != nullptr) memcpy(outDequeueSize, &removedSize, sizeof(unsigned int));
    return true;

}