Send,Recv Buffer

SendBuffer 구조

 

Buffer에 만들려는 패킷의 크기만큼  SendBufferManager::Open을 통해 원하는 크기의 SendBuffer의 Buffer를 가져온다. SendBuffer의 Buffer는 오브젝트 풀에서 생성될 때 SendBufferChunk의 Buffer로 초기화 된다.

(SendBuffer의 _buffer == SendBufferChunk의 _buffer)

(xarray<BYTE, SEND_BUFFER_CHUNK_SIZE>_buffer = {}; 에서 _usedSize 부터 시작하는 주소를 반환.)

 

가져온 SendBuffer의 Buffer의 메모리 공간을 PacketHeader로 캐스팅해서 패킷헤더로 만들어준다.

그리고 패킷헤더의 다음 주소는 직렬화된 패킷의 데이터를 저장한다.

 

SendBuffer의 Buffer에 패킷의 정보가 만들어졌다면, Close 함수를 호출한다.

패킷의 크기만큼 사용했기 때문의 _writeSize를 패킷 크기만큼 늘려준다.

마지막으로 SendBufferChunk의 Close도 호출해서 _open, _usedSize 변수 값을 설정해준다.

 

이렇게 패킷의 정보가 담긴 SendBuffer를 만들고 리턴해줘서 이를 이용해 서버 또는 클라이언트로 전송한다.

 

수정 전 Send (SendBuffer 사용 전)

SendEvent* sendEvent = xnew<SendEvent>();
sendEvent->owner = shared_from_this(); //ADD_REF
sendEvent->buffer.resize(len);
memcpy(sendEvent->buffer.data(), buffer, len);

WRITE_LOCK;
RegisterSend(sendEvent);

 

수정전에 3가지 문제점이 있었다.

 

첫 번째로, 복사 비용이 많이 든다는 점이다. 

1명 한테만 데이터를 전송하는 경우는 괜찮지만, mmorpg에서는 다른 클라이언트에게 브로드 캐스팅을 해주는 경우가 빈번하다. 이런 경우 유저의 수 만큼 데이터를 보낼 때 마다 데이터를 복사하게 된다.

 

두 번째로,

데이터를 모아서 보낼 수 있음에도 그러지 않고 매번 보내고 있다.
그래서 Send를 할 때마다 RegisterSend()를 해주기 보다는 데이터를 세션쪽에 모아놨다가 한번에 보내도록 해줘야 한다.

 

세 번째로, 데이터를 보낼 때 마다 SendEvent를 새로 생성해주고 삭제한다.

 

위 문제를 해결하기 위해서는 공용으로 사용할 SendBuffer를 만들고 이 SendBuffer 또한  Session 처럼 레퍼런스 카운팅을 통해 관리해주어야 한다.

iocp에 등록해서 Dispatch 함수를 통해 복원시켜주기 위해 Send를 하고 완료통지를 받아서 ProcessSend()가 호출될 때 까지 Session을 레퍼런스 카운팅을 통해 관리하는 것 처럼, SendBuffer 또한 레퍼런스 카운팅을 통해 WSASend가 호출완료 될 때 까지 유지 시켜주어야 한다.

그리고 SendEvent를 매번 새로 만들어서 보낼 데이터가 생길 때 마다 보내는게 아니라,

SendEvent를 하나 만들어 놓고 데이터를 모았다가 전송하고 GQCS를 통해 전송이 완료되면 만들었던 SendEvent를 재상용 하도록 해줄것이다.

 

 

SendBuffer 클래스를 이용하도록 Session에 SendBuffer Queue를 만들어 주었다.

 

 

SendEvent에도 SendBuffer를 저장할 vector 생성.

 

 

 

 

왜 SendEvent에서도 SendBuffer를 저장할 수 있게 해줄까?

Session에서 이미 SendBuffer Queue로 보낼 데이터를 저장하고 있는데 말이다.

이유는 이렇다.

WSASend를 호출하는 순간 완료될 때까지 SendBuffer가 절대로 없이지지 않게끔 레퍼런스 카운팅을 유지시켜줘야 한다.

하지만 queue에서 데이터를 pop하는 순간 레퍼런스 카운트가 0이 되면서 사라질 수도 있기 때문에,

이를 방지하기 위해 SendEvent에서 한번 더 보관해주도록 하였다.

(혹시나 코드가 변경되면서 RegisterSend가 락을 잡지 않고 호출될 수도 있으니, 락을 한번 더 잡아줬다.)

 

SendEvent로 데이터를 옮겼다면 이제부터는 멀티스레드 환경을 신경 쓸 필요가 없다.

한 개의 스레드만 호출하기 때문이다.

 

이 후, 성공적으로 WSASend가 호출이 되어서 GQCS를 통해 ProcessSend()가 호출된다.

이 때는 SendEvent의 값들을 날려줘서 레퍼런스 카운트를 줄여준다.

그리고 락을 잡은 후, 현재 SendBuffer가 비어있는지 확인 후, 비어있을 경우 _sendEvent를 false로 만들어서 다른 스레드가 RegisterSend를 호출할 수 있도록 해준다. 만약 SendBuffer가 비어있지 않았다면 다시 RegisterSend를 호출한다.

 

SendBuffer가 사용되는 순서.

    SendBufferRef sendBuffer = GSendBufferManager->Open(packetSize);
    PacketHeader* header = reinterpret_cast<PacketHeader*>(sendBuffer->Buffer());

 

1.PacketHandler를 통해 패킷이 SendBuffer의 Buffer에 생성됨.

 

2. Session::Send() 또는 Room::BroadCasting() 호출

_sendQueue에 보낼 버퍼를 저장한다.

그리고 그 중 한개의 스레드만 RegisterSend() 호출.

 

3. Session:: RegisterSend() 

_sendQueue가 빌때까지 pop()을 해서 _sendEvent의 sendBuffers로 옮긴다.

데이터를 모아서 전송하기 위해 xvector<WSABUF> wsaBufs;를 만들어준다.

_sendEvent의 sendBuffers를 for each문으로 접근하여 wsaBufs를 채워준다.

채워진 wsaBufs를 WSASend를 통해 전송한다.

 

 

일단 send()에서 _sendQueue에 버퍼를 넣어주고 _sendRegistered를 true 변경후  RegisterSend()를 호출합니다. 그리고 _sendQueue를 모두 비운후 WSASend까지 성공을하면 send()에서 걸었던 lock이 해제가 됩니다.

그후에는 다른 쓰레드에서 동일한 세션으로 send를 요청할경우 send()를 받고 _sendQueue에 버플 넣을수는 있으나 ProcessSend()에서 _sendRegistered를 false로 변경하지않았을경우 send()에서 RegisterSend()를 호출할수 없습니다. 그후 기존에 send가 진행되고 있던 쓰레드가 실행될경우 ProcessSend()에서 바로 RegisterSend()함수를 호출하게 되고 그때는 _sendQueue에 쌓여있는 버퍼를 볼수있게 됩니다.

 

버퍼는 절대 건드리면 안 되고 유지시켜줘야 하지만, wsabuf는 상관없습니다. (wsabuf.buf가 중요)
힙영역에 올라와 있는 버퍼를 쪼개서 sendBuffer로 사용하고, 입출력 완료까지 유지되도록 RefCount 관리를 해주고 있기 때문에 버퍼의 안전성이 보장된다.

---

RecvBuffer 구조

RecvBuffer는 SendBuffer에 비해 간단하다.

RecvBuffer를 Session에서 생성해서 이를 사용하는게 전부이다.

 

RecvBuffer는 데이터를 readPos와 wirtePos로 읽는다.

WSARecv로 데이터를 받아와서 ProcessRecv가 호출되면 wirtePos를 받아온 데이터 만큼 증가시킨다.

그리고 데이터를 읽고 나서는 readPos를 데이터 크기 만큼 증가시킨다.

 

우선 recv같은 경우에는 멀티쓰레드 환경에서도 안전하다.

왜냐하면 한번에 한 쓰레드만 RegisterRecv()와 ProcessRecv()로 들어오도록 코드를 작성하였기 때문이다.

 

기존 _recvBuffer는 일반 포인터를 사용하고 있었다.

그리고 TCP 특성상 100바이트를 보냈다고 한번에 100바이트를 받는게 아니라 나눠서 받게 되는 경우가 생길 수 도 있다.

하지만 수정하기 전에는 Session에 RegisterRecv()는 항상 데이터가 처리 되었다고 가정하고 시작주소를 건네주어서 기존의 정보를 덮어 씌우게 된다.

 

이런 현상을 해결하기 위해서는 패킷이 완전체로 왔는지 판별해줘야 한다.

그래서 패킷이 완전하게 오지 않았을 경우에는 기존에 있던 데이터를 남기고 남은 데이터를 덧 붙여 줘야한다.

 

버퍼 크기에 10을 곱한이유는?

복사비용을 줄이기 위해서이다.
우리가 사용하는 최대 버퍼 크기보다 더 크게 잡아주는 것이다.
예를 들어 우리가 버퍼의 최대크기를 5바이트로 잡아 놓았다면,
아래와 같이 한참 더 크게 만들어 놓고, 여기서 계속 read와 write를 하다보면
언젠간 read와 write가 겹치게 되고 그렇게 되면 복사비용 없어지면서 최적화가 될 것이다.
[   ][   ][   ][   ][   ]      [   ][   ][   ][   ][   ]      [   ][   ][   ][   ][   ]      [   ][   ][   ][   ][   ]      [   ][   ][   ][   ][   ]

 

(물론 writePos가 버퍼의 끝까지 갔는데 아직 데이터 남아있다면 어쩔 수 없이 데이터를 버퍼의 처음 위치로 복사해 와야 할 것이다.)

 

즉, 최대한 복사가 일어나지 않도록 해주기 위해 10을 곱해서 최대 버퍼 크기보다 더 크게 잡아주었다.

 

RecvBuffer 가 사용되는 순서.

 

1.Session 생성자에서 _recvBuffer 생성

 

 

2. WSARecv 호출. 

 

WSABUF에 _recvBuffer의  writePos로 시작하는 메모리 주소를 넘겨줌.

recv한 데이터가 _recvBuffer에 저장된다.

 

 

3. ProcessRecv 호출

성공적으로 WSARecv가 호출되고 나면 GQCS를 통해 ProcessRecv가 호출된다.

recv한 데이터의 크기만큼 writePos를 증가시켜준다.

그 후, OnRecv 함수를 호출하여 패킷을 조립한다. 

 

 

void RecvBuffer::Clean()

 

우선 데이터 크기를 가져온다. (writePos-readPos)

 

만약 데이터 크기가 0일 경우, 읽기+쓰기 위치가 동일하다는 소리이기 때문에 둘 다 시작위치로 리셋시킨다.

(복사 비용 없음)

 

하지만 데이터 크기가 0이 아니고 여유 공간(데이터 작성 가능한 공간)이 버퍼 1개 크기 미만이면,

남은 데이터를 시작위치로 복사한다. (데이터를 앞으로 당김)

기존에 버퍼에 남아있는 데이터는 삭제할 필요 없다. (어차피 덮어쓸거기 때문)

 

4. OnRecv 함수에서 패킷 조립

 

받아온 데이터로 패킷 헤더를 파싱한다.

해당 패킷 헤더를 통해 패킷의 크기와 받은 데이터의 크기를 비교해서 패킷 조립이 성공했는지 확인한다.

성공했을 경우에는 해당 패킷을 처리하는 함수를 ClientPacketHandler에서 호출해주고, OnRecv 함수를 빠져나온다.

실패 했을 경우에는 그냥 OnRecv 함수를 빠져나온다.

그 후, _recvBuffer.Clean(); 함수를 호출해서 커서를 정리하고 다시 데이터를 recv 할 수 있도록 RegisterRecv를 걸어놓고 빠져나온다.

 

즉, 기존의 버퍼를 날리는게 아니라 계속 누적한 채로 동작한다.