1 큐잉의 목적 #

큐는 자료구조 특성상 FIFO(First In First Out)의 구조를 가진다. 자료의 처리 순서를 흐트리지 않고 일시적으로 쌓기 위해서 사용되는데,

1. 소켓프로그래밍에서는 들어온 자료를 일단 읽어서 쌓고, 시간이 걸리는 자료분석 및 처리는 뒤로 하기 위해서 (이렇게 함으로써 최고의 소켓 효율을 이끌어낸다) 사용한다. (Read Buffering)

2. WOULDBLOCK : send 혹은 write 에서 EWOULDBLOCK이 발생하는 경우, 즉 운영체제의 내부 소켓버퍼가 가득차서 일시적으로 데이터 송신을 할 수 없는 경우를 위해 사용한다. 일시적이기 때문에 에러라고 생각해서 연결을 끊어버릴 수도 없고, 그렇다고 못보내고 남은 데이터를 따로 보관해서 처리하기도 번거롭다.

3. Small Packet : 1바이트를 보내기위한 패킷오버헤드는 약 64배에 달한다. 네트워크 대역폭이 100Mbps 의 Fast Ethernet 이라고 할때에, 1바이트씩의 패킷만을 연달아 송신하면 겨우 1.6Mbps의 자료를 전송할 수 있다. 운영체제는 이 문제를 피해가기 위해 내부 알고리즘을 이용해서 최대한 데이터를 묶어서 전송한다. 어플리케이션 차원에서도 작은 데이터들을 묶어서 최대한 전송하기 위해 Send Buffering을 이용한다.

2 큐잉의 구현 #

2.1 STL queue를 이용한 구현 #

2.1.1 Text Data Queuing 구현 #

#include <iostream>
#include <queue>

using namepsace std;

int main(void)
{
    queue<char *> binq;

    char *p_pop;

    binq.push("Hello ");
    binq.push("World\n");

    while(!binq.empty()){
        cout << binq.front();
        binq.pop();
    }
}

queue 탬플릿의 push, front, pop 을 이용해서 자료를 넣고 빼고 출력을 해 보았다. 텍스트의 NULL Terminator 처리가 조금 미흡하지만, 이것으로도 괜찮다. 다만, 통신에 사용할 자료들이 바이너리 데이터라면(즉 0, NULL을 포함하는 것이라면) 적절치 않으므로 약간의 보완을 필요로 한다.

2.1.2 Binary Data Queuing 구현 #

#include <queue>
#include <iostream>

using namespace std;

class TBinData
{
public :
    int  size;
    char *buffer;

    TBinData(const int nsize, const char* buf)
    {
        size = nsize;

        buffer = new char[nsize];
        memcpy(buffer, buf, nsize);
    };

    ~TBinData()
    {
        delete []buffer;
    }
}; 

int main(void)
{
    queue<TBinData *> binq;

    TBinData Data1(12, "Hello World\n");
    TBinData Data2(12, "Bin Queuing\n");

    binq.push(&Data1);
    binq.push(&Data2);

    char buf[1024];

    while(!binq.empty()){
        TBinData *data = binq.front();
        memcpy(buf, data->buffer, data->size);

        // 테스트 출력을 위해 터미네이터 붙임
        buf[data->size] = 0;
        cout << buf;

        binq.pop();
    }

    return 0;
}

?TBinData라는 클래스를 만들어, 이 클래스의 포인터들을 큐로 관리하도록 하였다. 이렇게 해 놓고 보니 만약 push되는 데이터들이 모두 1 바이트짜리라면.. 메모리에 많은 낭비가 생길듯하다. 어차피 바이너리 스트림이라면, 여러개의 데이터를 하나로 합쳐놔도 문제는 없을 듯 하다.

2.1.3 큐의 개선 : 각 요소들의 크기를 적당한 사이즈로 만든다 #

적당한 사이즈란 과연 얼마일까? 여기저기 듣자하니 어쨌거나 1400 이라고들 한다. 이 1400 이라는 값은 그러니깐 한번에 소켓으로 쏘기에 적당한 값이라는 의미이다. 자료구조들도 이에 맞춰서 작성하면 좋을듯 하다.

class TBinData
{
public :
    int  size;
    char *buffer;

    TBinData(const int nsize, const char* buf)
    {
        size = nsize;

        buffer = new char[nsize];
        memcpy(buffer, buf, nsize);
    };

    ~TBinData()
    {
        delete []buffer;
    }

    int Add(const int nsize, const char *buf)
    {
        int newsize = size + nsize;

        // 새로운 사이즈로 할당
        char *tmpbuf = new char[newsize];

        // 두개의 데이타 복사
        memcpy(tmpbuf, buffer, size);
        memcpy(&tmpbuf[size], buf, nsize);

        // 기존버퍼 삭제후 새 버퍼 대입
        delete []buffer;
        buffer = tmpbuf;
        size   = newsize;

        return newsize;
    };
}; 

////

queue<TBinData *> binq;
TBinData *LastData = NULL;

int AddNew(const int size, const char *buffer)
{
    LastData = new TBinData(newsize, buffer);
    binq.put(LastData);

    return binq.size();
}

int main(void)
{
...
    if(binq.empty()){
       AddNew(size, buffer);
    } else {
       if(LastData->size + size <= 1400) LastData->Add(size, buffer);
       else AddNew(size, buffer);
    }
....
}

큐에 데이터를 추가할때에 이미 들어있는 데이터의 사이즈와 추가하려는 데이터의 사이즈의 합이 1400 이 넘지 않은 경우에는 큐의 요소를 늘리지 않고 그 값만 조정했고, 넘어가는 경우 혹은 아예 큐가 비어있는 경우는 큐 요소를 새로 추가하도록 하였다. TBinData:?:Add 함수내부적으로 조금 복잡하고 효율이 좋지 않다. 역시 1 바이트짜리 push 가 많으면 성능이 아주아주 좋지 않을 듯하다.

2.2 큐의 직접 구현 #

직접 구현은 이렇게 할 것이다. 적당량의 버퍼를 미리 잡아놓고 head 와 tail 포인터를 옮겨다니면서 push, pop 을 구현한다. 말은 쉽지만 역시 구현하기는 만만치 않다.

#include <string.h>
#include <queue>
#include <iostream>

using namespace std;

#define MAX_BUF 81920

class TBinQueue
{
protected :
    char buf[MAX_BUF];
    int  head, tail;

public :
    TBinQueue(void) : head(0), tail(0) {};
    int Push(const int nsize, const char *buffer);
    int Pop(const int maxsize, char *outbuffer);
    int View(const int maxsize, char *outbuffer) const;
    int Skip(const int skipsize);
};

int TBinQueue::Push(const int nsize, const char *buffer)
{
    if(tail + nsize >= MAX_BUF) return -1; // overflow

    memcpy(&buf[tail], buffer, nsize);
    tail += nsize;

    return (tail - head); // stored data size
}

int TBinQueue::Skip(const int skipsize)
{
    head += skipsize;

    if(head == tail){ 
        head = tail = 0;
    }
    
    return (tail - head);
}

int TBinQueue::Pop(const int maxsize, char *outbuffer)
{   
    int viewsize = View(maxsize, outbuffer);
    Skip(viewsize); // recalc head & tail
    
    return viewsize;
}

int TBinQueue::View(const int maxsize, char *outbuffer) const
{   
    int copysize = maxsize;
    int datasize = (tail - head);
    
    if( datasize < maxsize ) copysize = datasize;
    if( copysize < 1) return 0;
    
    memcpy(outbuffer, buf, copysize);
    
    return copysize;
}

int main(void)
{
    TBinQueue binq;
    char buffer[MAX_BUF];
    
    binq.Push(13, "Hello World.\n");
    binq.Push(13, "Bin Queuing.\n");
    
    int nsize = binq.View(MAX_BUF, buffer);
    binq.Skip(nsize);
    
    buffer[nsize] = 0; // termination
    cout << buffer;
    
    return 0;
}