I/O Multiplexing epoll 채팅서버 만들기 with.C/C++

개발환경은 도커 debian trixie 컨테이너 입니다.

ai의 도움을 받아 공부하며 작성된 게시글입니다.

---

I/O Multiplexing

여러 파일 디스크립터를 동시에 감시하여, 블로킹 없이 준비된(바로 실행 가능한) I/O만 처리할 수 있도록 하는 기술이다. 특히 데이터 도착 시점을 예측할 수 없어서 준비됐는지 아닌지 모르는 네트워크 소켓이나 파이프와 같은 이벤트 기반 I/O에 주로 사용된다.

poll

poll은 여러 FD를 등록해놓고 읽기/쓰기 가능한 상태가 된 것을 알려주는 시스템 콜이다.

select랑 거의 똑같이 쓸 수 있다.

int poll(struct pollfd* fds, nfds_t nfds, int timeout);

• fds : 감시할 fd 배열
• nfds : 배열 길이
• timeout : -1 무한대기, 0 바로 리턴, ms단위 시간

struct pollfd {
	int fd; //감시 대상
    short events; //내가 감시하고 싶은 이벤트
    short revents; // 실제로 발생한 이벤트
};

주요 event 목록(전체는 더 많다)

• POLLIN : 읽기 가능
• POLLOUT : 쓰기 가능
• POLLERR : 에러 (등록 안해도 됨)
• POLLHUP : 연결 종료 (등록 안해도 됨)

poll은 내부적으로 이렇게 동작한다.

1. 유저가 fd배열을 커널에 넘김
2. 커널이 배열을 하나씩 검사함
3. 이벤트 발생한 fd 를 표시
4. 다시 유저에게 돌려줌

매 poll함수를 불러올때마다 배열 전체를 순회하기때문에 O(n)의 시간복잡도를 갖는다. 때문에 접속자가 너무 많아지면 좋지 않다.

select와 거의 똑같은데, 다른점은 등록가능 fd 수에 제한이 없다는 점, bitmask가 아니고 배열이라 확장성이 더 좋다는 점이 다르다.

poll의 실습은 넘어가자.

---

epoll 이란?

epoll은 리눅스에서 제공하는 고성능 I/O Multiplexing 시스템 콜이다.

poll이 모든 FD를 순회하는 것과 달리 이벤트가 발생한 FD만 반환하기 때문에 시간복잡도가 O(1)에 가깝다.

이게 가능한 이유는 커널이 FD를 등록해놓고 이벤트 발생 시 내부 ready 리스트에 넣어두기 때문.

 

epoll의 사용흐름은 이렇다.

1. 감시용 FD 생성
2. 감시할 FD 등록
3. 이벤트 대기

epoll은 두가지 모드가 있다

• LT  (기본값)
  - 읽기/쓰기 가능하면 계속 알림
• ET
  - 상태가 변화할 때만 알림 (읽기/쓰기 불가능 -> 읽기/쓰기 가능)
    그래서 반드시 while read로 recv buffer 비울때까지 돌려야함.
    그런데 다읽고나서 한번 더 read했을 때 blocking이면 거기서 멈춰버리니 소켓은 non-blocking이어야 함
    non-blocking이면 읽을거 없으면 바로 -1 return하고, errorno가 EAGAIN이면 '더 읽을게 없구나' 하는것.
  - 데이터 100바이트 도착 -> 1번 알림
  - 50바이트만 읽고 50바이트 남아있으면? 읽기 가능->읽기 가능으로 상태변화 없으니 알림x

더 깊게

- epoll 내부구조

epoll 객체 (epfd가 가리키는 커널안 객체)

ㄴ 레드블랙트리 - 등록된 FD 저장

ㄴ ready list - 이벤트 발생한 FD들 저장

 

epoll_ctl(ADD) 할 때 이런 작업이 일어난다.

1. RB-tree에 FD등록(FD 관리용)

2. 해당 소켓의 wait queue에 epoll callback 연결

 

그리고 이벤트(네트워크 패킷 도착)발생 시 이렇게 흘러감

1. 커널이 소켓 recv buffer 에 기록

2. 소켓의 wait queue 깨움

3. 거기에 등록된 epoll_callback 실행됨

4. callback 안에서 해당FD를 ready list에 추가

c/c++ 로 epoll 채팅서버 만들기

수도코드

accept 소켓 설정하기
accept 소켓 epoll에 등록하기
while(1)
	epoll_wait 대기
	대기 풀리면 소켓 리스트 받음
	for each 소켓 리스트
		소켓이 accept 신호로 깨어난거면
			새 클라이언트 접속 있다.
			while accept 다할때까지 클라이언트 받고 epoll에 등록
		소켓이 read 신호로 깨어난거면
			어떤 클라가 메세지 보냈다.
			모든 클라이언트 소켓에 메세지 write

epoll_event 구조체

• epoll에 fd 감시 등록할 때 epoll_event 객체를 epoll_ctl 함수의 인자로 넣는다.
• events 는 이 소켓에서 내가 감지할 이벤트를 설정하는 것. 복수로 설정하고싶다면 이런식으로 넣자 EPOLLIN | EPOLLOUT
• events를 EPOLLOUT로 등록해놓으면 쉬지않고 계속 감지될 것이므로 쓸게 있을때만 EPOLLOUT를 추가해서 설정하고, 다 쓰면 EPOLLIN만 설정해놓도록 하자.
• epoll_wait로 대기하다가 이벤트가 발생하면 epoll_event 배열을 준다. 이중 events는 이전에 감지할 이벤트로 설정한 것 중에 어떤 이벤트를 감지했는지 알려준다.
• data는 등록할 때 우리가 넣었던 값이다. ptr을 잘 써먹으면 좋을 것 같다.

struct epoll_event {
   uint32_t      events;  /* Epoll events */
   epoll_data_t  data;    /* User data variable */
};

union epoll_data {
   void     *ptr;
   int       fd;
   uint32_t  u32;
   uint64_t  u64;
};

typedef union epoll_data  epoll_data_t;

소스코드

#include <iostream>
#include <vector>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <string>
#include <sys/epoll.h>
#include <stdlib.h>

using namespace std;

void	handle_error(const char* err_str, int rtn);
int		open_listen_socket(int port);
void	make_socket_nonblock(int sock);
void	broadcast(int fd, const vector<int>& c_sockets, const char* buffer, int size);
void	erase_from_vector(int fd, vector<int>& vec);

int main(int ac, char** av)
{
	if (ac != 2)
		handle_error((string(av[0]) + " [port]").c_str(), 1);
	int listen_sock = open_listen_socket(atoi(av[1])); //listen socket 생성

//        ##### epoll 설정 단계 #####
	struct epoll_event ev;
	ev.data.fd = listen_sock;
	ev.events = EPOLLIN;
	int epollfd = epoll_create1(0);
	if (epollfd == -1)
		handle_error("epoll_create1 error", 1);
	if (epoll_ctl(epollfd, EPOLL_CTL_ADD, listen_sock, &ev) == -1)
		handle_error("epoll_ctl listen_sock error", 1);
	vector<struct epoll_event>	events(1);
	vector<int>					c_sockets;
  
//        ##### 채팅 서버 시작 #####
	while (true)
	{
		int nfds = epoll_wait(epollfd, events.data(), events.size(), -1);
		for (int n = 0; n < nfds; n++)
		{
			if (events[n].data.fd == listen_sock)
			{// 새 클라이언트 connect
				struct sockaddr_in client_addr;
				socklen_t sock_len = sizeof(struct sockaddr_in);
				int client_sock = accept(listen_sock, (struct sockaddr*)&client_addr, &sock_len);
				if (client_sock == -1) {
					if (errno == EAGAIN)
						break;
					handle_error("accept error", 1);
				}
				events.push_back({});

				make_socket_nonblock(client_sock);
				ev.events = EPOLLIN;
				ev.data.fd = client_sock;
				if (epoll_ctl(epollfd, EPOLL_CTL_ADD, client_sock, &ev) == -1)
					handle_error("epoll_ctl client add error", 1);
				c_sockets.push_back(client_sock);
			} else
			{ // 클라이언트가 서버로 메세지 보냄
				char read_buffer[100];
				int rd = read(events[n].data.fd, read_buffer, 100);
				if (rd == -1)
				{
					if (errno == EAGAIN || errno == EWOULDBLOCK)
						continue;
					handle_error("read error", 1);
				}
				if (rd == 0)
				{ // 클라이언트가 나감
					int fd = events[n].data.fd;
					erase_from_vector(fd, c_sockets);
					epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, NULL);
					close(events[n].data.fd);
					broadcast(fd, c_sockets, "out\n", 4);
					events.pop_back();
					continue;
				}
				broadcast(events[n].data.fd, c_sockets, read_buffer, rd);
			}
		}
	}
}

void handle_error(const char* err_str, int rtn)
{
	cerr << err_str << endl;
	exit(rtn);
}

int open_listen_socket(int port)
{
	int listen_sock = socket(AF_INET, SOCK_STREAM, 0);
	if (listen_sock == -1)
		handle_error("socket error", 1);
	make_socket_nonblock(listen_sock);
	struct sockaddr_in sock_addr;
	memset(&sock_addr, 0, sizeof(sock_addr));
	sock_addr.sin_family = AF_INET;
	sock_addr.sin_port = htons(port);
	sock_addr.sin_addr.s_addr = INADDR_ANY;

	if (bind(listen_sock, (struct sockaddr *)&sock_addr, sizeof(sock_addr)) == -1)
		handle_error("bind error", 1);
	if (listen(listen_sock, 10) == -1)
		handle_error("listen error", 1);
	return listen_sock;
}

void erase_from_vector(int fd, vector<int>& vec)
{
	for (vector<int>::iterator it = vec.begin(); it != vec.end(); it++)
	{
		if (*it == fd)
		{
			vec.erase(it);
			return ;
		}
	}
}

void make_socket_nonblock(int sock)
{
	int flags = fcntl(sock, F_GETFL, 0);
	if (flags == -1)
		handle_error("fcntl F_GETFL error", 1);
	if (fcntl(sock, F_SETFL, flags | O_NONBLOCK) == -1)
		handle_error("fcntl NONBLOCK error", 1);
}

void broadcast(int fd, const vector<int>& c_sockets, const char* buffer, int size)
{
	string str;
	str = "client " + to_string(fd) + ": " + string(buffer, size);
	for (int i = 0; i < c_sockets.size(); i++)
		write(c_sockets[i], str.c_str(), str.length());
}

 

• epoll_wait에 넣는 struct epoll_event 배열을 가변적길이로 사용하고싶어서 vector로 만들고 클라가 추가되면 길이 +1, 나가면 -1 되도록 만들었다. 늘이고 줄이고 할때마다 코드를 똑바로 써줘야해서 실수의 여지가 많다. 제대로 쓰려면 epoll관리자 클래스가 하나 필요할 듯 하다. 아니면 최대치를 고정한다던지 다른방법을 쓰거나.
• write는 비동기를 신경쓰지 않았다. 대부분의 경우 커널의 send buffer에 다 쓸 수있을거라 생각하기 때문. 하지만 만약 write 시도 중 커널의 send buffer가 꽉차서 일부만 write 한다면 보내지 못한 데이터는 날아가버린다. 그러므로 제대로 만들거라면 send buffer를 세션마다 가지게 하고 write하고 남은 부분을 다음에 write가 가능할 때 보낼 수 있도록 만들어야 한다.

ncat을 사용해서 접속해보자. (없으면 apt install ncat)

ncat localhost port

14개의 클라를 연결해봤는데 잘 된다.

다만 컨트롤c로 퇴장할 때는 서버측에서 확인이 되는데 창닫기로 나가면 확인이 안된다.