Kqueue로 구현해보는 I/O Multiplexing
이벤트 루프를 통해 단일 스레드로 여러 클라이언트 요청 처리하기
2024-11-04
while (true) {
const std::vector<event> events = get_events();
while (auto &event : events) {
if (event == READ_EVENT) {
read_handler(event);
} else if (event == WRITE_EVENT) {
write_handler(event);
}
}
}get_events() 부분에서 epoll() 등의 시스템 콜을 사용하게 됩니다.
select(), poll(), epoll() 등이 있는데요.select()와 poll()은 파일 디스크립터들의 상태를 커널(Kernel)이 아닌 유저 스페이스(User Space)에서 관리하므로 커널과 유저 스페이스 간 불필요한 통신이 발생하게 되는데요.epoll() 등의 시스템 콜은 커널에서 직접 파일 디스크립터들의 상태를 관리하므로 성능 부분에서 효율적이므로 일반적으로 epoll()을 사용해 I/O 멀티플렉싱을 구현하게 됩니다.
epoll()을 사용할 수 없었습니다.epoll()인 kqueue()를 통해 이벤트 루프를 구현해 보겠습니다.
struct kevent {
uintptr_t ident;
int16_t filter;
uint16_t flags;
uint32_t fflags;
intptr_t data;
void *udata;
};kevent는 파일 디스크립터인 ident, 이벤트를 선별하기 위해 사용하는 filter, 새로 등록할 이벤트에 대해 수행할 작업인 flags 등을 가지고 있습니다.ident를 해당 이벤트가 발생한 파일 디스크립터가 어떤 소켓의 파일 디스크립터인지 구별하기 위해 사용하게 됩니다.filter를 사용합니다.
int kevent(int kq,
const struct kevent *changelist,
int nchanges,
struct kevent *eventlist,
int nevents,
const struct timespec *timeout);kevent()입니다.changelist로 전달하며, 이미 Kqueue에서 관리하는 파일 디스크립터들로부터 발생한 이벤트들은 eventlist에 전달하게 됩니다.
#define EV_SET(kevp, a, b, c, d, e, f) do { \
struct kevent *__kevp__ = (kevp); \
__kevp__->ident = (a); \
__kevp__->filter = (b); \
__kevp__->flags = (c); \
__kevp__->fflags = (d); \
__kevp__->data = (e); \
__kevp__->udata = (f); \
} while(0)kevent는 직접 구조체를 생성해도 되지만, EV_SET()이라는 매크로 함수를 통해 편리하게 생성할 수 있습니다.
kqueue()를 통해 이벤트 루프를 구현해 보겠습니다.
class EventLoop {
const int &server_socket;
const sockaddr_in &server_address;
const socklen_t &address_len;
public:
EventLoop(const int &server_socket, const sockaddr_in &server_address, const socklen_t &address_len): server_socket(server_socket), server_address(server_address), address_len(address_len) {}
int run() const;
};EventLoop의 run()을 구현해보겠습니다.
EV_SET(&changed_event, server_socket, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, NULL);
change_events.push_back(changed_event);EV_SET()을 통해 서버 소켓에 대한 읽기 작업을 감지하는 이벤트를 생성해야 하는데요.kevent()의 changelist에 전달됩니다.
kevent(kq, change_events.data(), static_cast<int>(change_events.size()), events, MAX_EVENT_COUNT, nullptr)kevent()를 통해 change_events 내의 이벤트들을 Kqueue에 등록하고, 발생한 이벤트들을 events로 전달 받습니다.kevent()는 timeout이 없다면 임의의 이벤트가 발생할때까지 무한정 대기하게 됩니다.kevent()는 즉시 해당 이벤트를 반환하게 됩니다.
for (const auto &event: events) {
if (event.flags & EV_ERROR) {
continue;
}
if (event.filter == EVFILT_READ) {
if (event.ident == server_socket) {
const int client_socket =
accept(
server_socket,
reinterpret_cast<struct sockaddr *>(const_cast<struct sockaddr_in *>(&server_address)),
const_cast<socklen_t *>(&address_len)
);
if (client_socket < 0) {
std::cerr << "accept() failed";
continue;
}
char client_address[address_len];
inet_ntop(AF_INET, &server_address.sin_addr, client_address, address_len);
std::cout << "New connection from " << client_address << ":" << ntohs(server_address.sin_port) << std::endl;
EV_SET(&changed_event, client_socket, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, NULL);
change_events.push_back(changed_event);
} else {
const int client_socket = static_cast<int>(event.ident);
char buffer[MAX_BUFFER_SIZE];
const long n = read(client_socket, buffer, MAX_BUFFER_SIZE);
if (n <= 0) {
std::cout << "Client disconnected, fd: " << client_socket << std::endl;
close(client_socket);
} else {
std::cout << "Received: " << buffer << std::endl;
write(client_socket, buffer, n);
}
}
}
}kevent()가 발생한 이벤트들을 반환한 후, 이벤트 루프는 해당 이벤트들에 맞는 처리를 수행합니다.kevent의 ident와 flags를 사용하는데요.ìdent가 서버 소켓의 파일 디스크립터이고 flags가 읽기 작업에 대한 EVFILT_READ라면 서버 소켓이 연결 요청을 받았다는 것을 의미합니다.accept()를 호출해 새로운 클라이언트 소켓에 대한 파일 디스크립터를 받고 해당 파일 디스크립터에 대한 이벤트를 생성해 change_events에 저장합니다.kevent()의 changelist로 전달되어 Kqueue에 해당 이벤트에 대한 파일 디스크립터가 저장되게 됩니다.
flags가 EVFILT_READ인데 ident가 서버 소켓의 파일 디스크립터가 아니라면 클라이언트 소켓으로부터 데이터를 읽을 준비가 되었다는 것을 의미합니다.ident로부터 클라이언트 소켓을 가져오고 해당 소켓으로부터 read()와 write()를 통해 데이터를 읽고 클라이언트에게 그대로 전송하도록 했습니다.
int EventLoop::run() const {
int kq;
std::vector<struct kevent> change_events;
struct kevent events[MAX_EVENT_COUNT];
struct kevent changed_event{};
if (listen(server_socket, 5) < 0) {
std::cerr << "listen() failed";
return -1;
}
if ((kq = kqueue()) < 0) {
std::cerr << "kqueue() failed";
return -1;
}
EV_SET(&changed_event, server_socket, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, NULL);
change_events.push_back(changed_event);
std::cout << "Server listening on " << ntohs(server_address.sin_port) << std::endl;
while (true) {
if (kevent(kq, change_events.data(), static_cast<int>(change_events.size()), events, MAX_EVENT_COUNT, nullptr) == -1) {
std::cerr << "kevent() error";
break;
}
change_events.clear();
for (const auto &event: events) {
if (event.flags & EV_ERROR) {
continue;
}
if (event.filter == EVFILT_READ) {
if (event.ident == server_socket) {
const int client_socket =
accept(
server_socket,
reinterpret_cast<struct sockaddr *>(const_cast<struct sockaddr_in *>(&server_address)),
const_cast<socklen_t *>(&address_len)
);
if (client_socket < 0) {
std::cerr << "accept() failed";
continue;
}
char client_address[address_len];
inet_ntop(AF_INET, &server_address.sin_addr, client_address, address_len);
std::cout << "New connection from " << client_address << ":" << ntohs(server_address.sin_port) << std::endl;
EV_SET(&changed_event, client_socket, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, NULL);
change_events.push_back(changed_event);
} else {
const int client_socket = static_cast<int>(event.ident);
char buffer[MAX_BUFFER_SIZE];
const long n = read(client_socket, buffer, MAX_BUFFER_SIZE);
if (n <= 0) {
std::cout << "Client disconnected, fd: " << client_socket << std::endl;
close(client_socket);
} else {
std::cout << "Received: " << buffer << std::endl;
write(client_socket, buffer, n);
}
}
}
}
}
return 0;
}kqueue()를 통해 이벤트 루프를 구현했는데요.