버퍼링을 적용한 ?TEpollSocket 개선 #
엄밀히 말하면,
EpollSocket을 개선하는 것이 아니고
?TClient를 개선하는 것. 이름도 바꿔서
?TBasicClient 라고 하겠다. 물론 사용자는 이번 작업을 통해서 버퍼링에 대해서 신경쓰지 않아도 될 것이고, 데이터를 송신 & 수신하는 것까지 신경쓰지 않아도 된다.
- 1 버퍼링을 적용한 ?TEpollSocket 개선
- 1.1 원래의 ?TClient
- 1.2 개선하려는 ?TBasicClient
- 1.3 ?OnRead()의 구현
- 1.4 운영체제 소켓버퍼 조정
- 1.5 ?OnWrite() 구현
- 1.6 Flush() 구현
- 1.7 ?TBasicClient 동작 정리
- 1.8 기타 구현
- 1.9 소스
1.1 원래의 ?TClient #
#include "epsocket.h"
class TClient : public TEpollClient
{
public :
TClient(const int lfd, const char *laddr) : TEpollClient(lfd, laddr)
{
// 초기화때 추가적으로 해주고싶은 일들이 있다면 여기에 ...
};
virtual int OnRead(void);
virtual int OnWrite(void);
virtual int OnError(const int nerror);
};
1.2 개선하려는 ?TBasicClient #
#include "epsocket.h"
#include "buffer.h"
#define MAX_BUF (128 * 1024) // 128K
class TBasicClient : public TEpollClient
{
protected :
TBinQueue *sendbuf, *recvbuf;
bool is_would_block;
bool is_error;
public :
TBasicClient(const int lfd, const char *laddr) : TEpollClient(lfd, laddr)
{
// 초기화때 추가적으로 해주고싶은 일들이 있다면 여기에 ...
is_would_block = false;
is_error = false;
sendbuf = new TBinQueue(MAX_BUF);
recvbuf = new TBinQueue(MAX_BUF);
};
~TBasicClient()
{
delete sendbuf;
delete recvbuf;
};
virtual int OnRead(void);
virtual int OnWrite(void);
virtual int OnError(const int nerror);
virtual int OnIdle(void);
};
?TBinQueue의 버퍼로 sendbuf, recvbuf 두개를 가지고 있고 128K의 크기를 가지도록 하였다. 버퍼는 말 그대로 일시적인 완충의 목적이므로 작게 잡거나 크게 잡거나 큰 의미는 없다. 다만 일시적인 서버부하 (예를 들어 디스크 입출력)를 감당할 만큼의 완충영역을 가지고 있어야 하므로, 충분히 잡아주는게 좋다. 하나의 커넥션은 128K 씩의 두개버퍼를 가지게되는데 (256K), 동시접속 1천이면 256MB의 메모리를 차지하게 된다. 1만을 커버하려면 2.5G의 버퍼메모리가 필요하니 -_-; 이 숫자를 줄이던가 메모리를 많이 사던가 하라!
동시접속이 10 이내인 다계층 서버 (즉, 이용자 접속은 Agent가 받고 게임서버와 채널 하나로 통신한다던가 하는...)라면 커넥션숫자보다는 커넥션당 전송량이 더 큰 의미가 있으므로 이 버퍼를 10M 이상으로 잡을 수도 있다.
버퍼사이즈를 사용자가 정하게 하도록 인터페이스를 수정해보았다. constructor는 epsocket이 호출하기때문에 간접적 구현을 했는데, 내부적으로 버퍼사이즈를 리턴해주는 함수를 작성하고, 사용자가 재정의 할 수 있도록 한다. 디폴트로는 128K 버퍼를 생성하도록 한다.
protected :
virtual int internalbuffersize(void) = { return MAX_BUF; };
...
public :
TBasicClient(...)
{
...
int maxbuf = internalbuffersize();
sendbuf = new TBinQueue(maxbuf);
recvbuf = new TBinQueue(maxbuf);
};
~TBasicClient()
{
delete sendbuf;
delete recvbuf;
};
...
1.3 ?OnRead()의 구현 #
int TBasicClient::OnRead(void)
{
char buf[SIZE_OSBUF];
int nread = read(fd, buf, SIZE_OSBUF);
if(nread == 0) return EP_ERROR_READ_CLOSE;
else if(nread < 0){
if(errno == EAGAIN) return EP_ERROR_NONE; // 에러는 아니며, 읽을 데이터가 없을 뿐
return EP_ERROR_READ_GEN;
}
if(recvbuf->Push(nread, buf) < 1) return EP_ERROR_READ_OVERFLOW;
nin += nread;
return EP_ERROR_NONE;
};
read 이벤트가 발생했을때, 즉 어떤 데이터가 수신되었을때
저절로 호출되는 핸들러의 구현이다. SIZE_OSBUF 상수는 setsockopt(..., SO_RCVBUF, ...) 에서 지정해준 값과 같게 하기 위해 사용된다. read의 세번째 파라미터 count에 대해서가 약간 애매모호한 구석이 있는데,
#include <unistd.h>
ssize_t read(int fd, void *buf, size_t count);
- 수신된 데이터가 count보다 많으면, count만큼 읽고(count와 같은값 리턴) 읽지 않은 나머지 데이터는 운영체제 버퍼에 남아있게 된다.
- 수신된 데이터가 count보다 같거나 작으면, 데이터사이즈만큼 읽고 그 사이즈를 리턴한다.
- 수신된 데이터가 없으면, 에러(-1)를 리턴하고 errno == EAGAIN 이 된다 (nonblocking 으로 만들어놨기때문이다. 만약 blocking socket이라면 여기서 데이터가 들올때까지 프로그램이 멈춘다 --);
- count의 값은 unistd.h 에 지정된 SSIZE_MAX 보다 작아야 한다. (크게 지정될 경우는 어떻게 될지 모른다 --; )
이런 이유로 애매모호함을 없애기 위해서, 최대로 읽어내야 하는 값과 운영체제소켓버퍼 사이즈를 같게 한다. 그러면 적어도 운영체제 버퍼에 데이터가 남는 일은 없게 된다.
1.4 운영체제 소켓버퍼 조정 #
#define SIZE_OSBUF 8192
int send_buf = SIZE_OSBUF;
int recv_buf = SIZE_OSBUF;
setsockopt(fd, IPPROTO_TCP, SO_SNDBUF, (char *)&send_buf, sizeof(send_buf));
setsockopt(fd, IPPROTO_TCP, SO_RCVBUF, (char *)&recv_buf, sizeof(recv_buf));
8K로 했는데, 프로그램이 자주데이터를 만들어내고 자주 비우는 경우라면 더 작게해도 상관없지만 한번에 많은 데이터를 쏘고 받고 하는 경우에는 더 크게 해주는게 좋다. 온라인게임서버들이라면 더 작게, ftp나 http서버라면 더 크게 한다. 하지만 그냥 8K면 여러목적에 무방할 듯하다.
1.5 ?OnWrite() 구현 #
int TBasicClient::OnWrite(void)
{
is_would_block = false;
return Flush();
}
?OnWrite()가 호출되었다는 것은 소켓이 쓰기가능하다는 의미인데, 1) 소켓이 처음 연결되었을때 2) WOULDBLOCK 상태가 해소되었을때 3) WOULDBLOCK이 아닌 상태에서, 데이터가 수신되었을때이다. 위 코드에서는 would_block flag를 false로 꺼주고
?SendBuffer를 Flush 하도록 작성하였다. 실제로 데이터 전송은 Flush() 에서 하게 된다.
1.6 Flush() 구현 #
#define SIZE_APACKET 1400
class TBasicClient : public TEpollClient
{
public :
int Flush(void);
....
};
....
int TBasicClient::Flush(void)
{
if(is_error) return EP_ERROR;
char buf[SIZE_APACKET];
int nsize;
while(! is_would_block){
if(sendbuf->empty()) return EP_ERROR_NONE;
nsize = sendbuf->View(SIZE_APACKET, buf);
int nwrite = write(fd, buf, nsize);
if( nwrite != nsize ){
// 뭔가 에러가 생긴것
if(nwrite < 1){
return OnError(EP_ERROR_WRITE_GEN);
}
is_would_block = true;
}
sendbuf->Skip(nwrite);
nout += nwrite;
}
return EP_ERROR_NONE;
}
?SendBuffer를 비울때까지 write 함수를 호출하여 실제 전송을 하는 함수이다. nin, nout 변수를 두어 받고 보낸 데이터양을 집계하도록 했다. 전송 도중에 에러가 발생하거나, WOULDBLOCK 상태가 되면 멈춘다. write 함수의 세번째 파라미터 count는 실제로 보내려고 하는 양인데, 함수의 결과값이 count와 같지 않은 경우는 1) 에러가 발생했거나( < 1 ), 2) WOULDBLOCK 이 된 경우이다. ( nwrite < nsize) WOULDBLOCK이 발생한경우에는 실제보낸 사이즈 nwrite 만큼만 버퍼를 비우고 Flush 함수수행을 중지한다.
다 보내지 못하고 남은 부분은, 다음번에
?OnWrite 가 호출되면서 다시 Flush 를 통해서 보내게 된다. 아니면 남는 시간에 Flush() 호출을 해준다 (필자는 이 방법을 선호한다)
?OnError()의 구현
int TBasicClient::OnError(const int nerror)
{
is_error = true;
return nerror;
}
특별히 하는 일은 없다. --;
?OnIdle()의 구현
int TBasicClient::OnIdle(void)
{
// 들어온 데이터 처리
while(! recvbuf->empty()){
char buf[SIZE_APACKET];
int nsize = recvbuf->Pop(SIZE_APACKET, buf);
if(sendbuf->Push(nsize, buf) < 1){
// cout << "sendbuffer overflow\n";
}
}
// Flushing
Flush();
return ERROR_NONE;
}
?OnIdle은,
?TEpollSocket에서 모든 연결에 대해 호출하는 함수 (아래에서 부연설명)로 메인쓰레드, 혹은 메인함수가 그냥 계속해서 호출해주는 함수이다. 데이터 수신&송신은 자동적으로
?OnRead,
?OnWrite에서 하지만 수신된 데이터의 분석&처리(=메시지 프로세싱)는 이 곳에서 하고 있다 (커넥션끼리의 상호커뮤니케이션이 많은 채팅서버를 제작할때에는 이 곳에서 하지 않고 클래스 외부에서 할 것이다).
프로그램은,
?RecvBuffer에서 적당한 사이즈의 패킷을 읽어서 처리 (echo서버이기 때문에 바로 다시 반송)하도록 하고 있다. 반송은 물론 직접 write 콜을 하지 않고 단지
?SendBuffer에 넣어주기만 한다. 실제 전송은
?OnWrite 를 통해 Flush가 할 것이다.
1.7 ?TBasicClient 동작 정리 #
여태까지를 대략 정리해보면,
- ?OnRead에서는 데이터를 읽어서 ?RecvBuffer에 담는다.
- ?OnWrite에서는 WOULDBLOCK flag를 셋한다.
- 남는시간에(?OnIdle) 들어온데이터의 분석&처리, 실제 데이터의 전송을 한다.
?OnIdle 구현
?OnIdle의 구현은, main 과
?TEpollSocket의 변경을 필요로 한다.
class TEpollSocket
{
....
public :
....
int Idle(void);
};
....
template <typename T>
int TEpollSocket<T>::Idle(void)
{
map<int, T *>::const_iterator it = mapClient.begin();
while(it != mapClient.end()){
it++ ->second->OnIdle();
}
return mapClient.size();
}
하는 일은 전체 연결(즉
?TBasicClient들)의 Idle을 호출해주는 일이다. for_each()로 구현하는 것이 멋질텐데 c++ 실력이 딸려서 컴파일되는 코드를 작성하는데 실패했다. 게다가 위 함수는 Typename Implicity 경고가 나온다. mapClient.begin() iterator의 타입이 무엇인지도 잘 모르겠다. --; 누가 좀 고쳐주세요.
컴파일러 경고 -_-;;
epsocket.h: In member function `int TEpollSocket<T>::Idle()':
epsocket.h:273: warning: `std::map<int, T*, std::less<int>,
std::allocator<std::pair<const int, T*> > >::const_iterator' is implicitly a typename
epsocket.h:273: warning: implicit typename is deprecated, please see the documentation for details
사실 이렇게 작성하고 싶지만... 능력이...
template <typename T>
int TEpollSocket<T>::Idle(void)
{
for_each(mapClient.begin(), mapClient.end(), mem_fun( ????? ::OnIdle));
}
main() 변경
#define WANT_MAX_CLIENT 10000
int main(void)
{
int n, max_client;
struct rlimit rlim;
getrlimit(RLIMIT_NOFILE, &rlim);
rlim.rlim_cur = WANT_MAX_CLIENT;
setrlimit(RLIMIT_NOFILE, &rlim);
getrlimit(RLIMIT_NOFILE, &rlim);
max_client = rlim.rlim_cur;
cout << "MAX_NOFILE : " << max_client << endl;
TEpollSocket<TBasicClient> esock(max_client);
esock.Listen(8000);
for(;;){
esock.Do(10);
esock.Idle();
}
}
esock.Do 와 esock.Idle을 무작정 호출하고 있다 --; getrlimit와 setrlimit로 열린파일갯수 조정을 하고 있는데, 필자 시스템에서 root유저로는 32768까지 가능하지만 일반유저권한으로는 1024가 한계였다.
Epoll Socket2