Linux 環境において高性能なネットワークサーバーを構築する際、I/O マルチプレクスングは不可欠な技術です。従来は epoll システムコールを直接制御して実装するのが一般的でしたが、libevent などのライブラリを利用することで、より簡潔なコードで同等の機能を実現できます。ここでは、単純なエコーサーバーを例に、raw epoll を使用した場合と libevent を使用した場合の実装差異を検証します。

1. raw epoll による実装

epoll を直接使用する場合、ファイルディスクリプタの登録、イベント待機、そして状態管理をすべて明示的に行う必要があります。以下はサーバー側とクライアント側の基本実装です。

サーバー側 (server_epoll.c)

ソケットの生成、バインド、リスニング後、epoll インスタンスを作成し、リスニングソケットを登録します。accept された新しい接続も再度 epoll に登録することで、複数クライアントを同時に処理します。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <fcntl.h>

#define MAX_EVENTS 10
#define BUFFER_SIZE 64

void set_nonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}

int create_listener_socket(int port) {
    int sock = socket(AF_INET, SOCK_STREAM, 0);
    if (sock < 0) return -1;

    int opt = 1;
    setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = INADDR_ANY;
    addr.sin_port = htons(port);

    if (bind(sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
        close(sock);
        return -1;
    }

    listen(sock, 10);
    set_nonblocking(sock);
    return sock;
}

void handle_client_message(int client_fd, int epfd) {
    char buf[BUFFER_SIZE];
    ssize_t n = read(client_fd, buf, sizeof(buf) - 1);
    
    if (n <= 0) {
        close(client_fd);
        return;
    }
    
    buf[n] = '\0';
    printf("Received: %s", buf);
    
    const char *msg = "ACK received\n";
    write(client_fd, msg, strlen(msg));
}

void run_event_loop(int listen_fd) {
    int epfd = epoll_create1(0);
    struct epoll_event ev, events[MAX_EVENTS];

    ev.events = EPOLLIN;
    ev.data.fd = listen_fd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev);

    while (1) {
        int nfds = epoll_wait(epfd, events, MAX_EVENTS, -1);
        for (int i = 0; i < nfds; i++) {
            if (events[i].data.fd == listen_fd) {
                struct sockaddr_in client_addr;
                socklen_t addr_len = sizeof(client_addr);
                int client_fd = accept(listen_fd, (struct sockaddr*)&client_addr, &addr_len);
                if (client_fd > 0) {
                    set_nonblocking(client_fd);
                    ev.events = EPOLLIN;
                    ev.data.fd = client_fd;
                    epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &ev);
                }
            } else {
                handle_client_message(events[i].data.fd, epfd);
            }
        }
    }
    close(epfd);
}

int main() {
    int port = 8080;
    int sock = create_listener_socket(port);
    if (sock < 0) {
        perror("Socket creation failed");
        return 1;
    }
    printf("Listening on port %d\n", port);
    run_event_loop(sock);
    return 0;
}

クライアント側 (client_epoll.c)

クライアント側でも標準入力とソケットの両方を epoll で監視し、非ブロッキング I/O を実現します。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <fcntl.h>

void set_nonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}

int connect_to_server(const char *ip, int port) {
    int sock = socket(AF_INET, SOCK_STREAM, 0);
    struct sockaddr_in addr;
    
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    inet_pton(AF_INET, ip, &addr.sin_addr);

    if (connect(sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
        close(sock);
        return -1;
    }
    set_nonblocking(sock);
    return sock;
}

void monitor_io(int sock_fd) {
    int epfd = epoll_create1(0);
    struct epoll_event ev, events[2];

    ev.events = EPOLLIN;
    ev.data.fd = STDIN_FILENO;
    epoll_ctl(epfd, EPOLL_CTL_ADD, STDIN_FILENO, &ev);

    ev.data.fd = sock_fd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, sock_fd, &ev);

    while (1) {
        int n = epoll_wait(epfd, events, 2, -1);
        for (int i = 0; i < n; i++) {
            if (events[i].data.fd == STDIN_FILENO) {
                char buf[256];
                ssize_t len = read(STDIN_FILENO, buf, sizeof(buf));
                if (len > 0) {
                    write(sock_fd, buf, len);
                } else {
                    return;
                }
            } else if (events[i].data.fd == sock_fd) {
                char buf[256];
                ssize_t len = read(sock_fd, buf, sizeof(buf));
                if (len > 0) {
                    fwrite(buf, 1, len, stdout);
                } else {
                    printf("\nServer disconnected\n");
                    return;
                }
            }
        }
    }
}

int main(int argc, char *argv[]) {
    if (argc != 3) {
        fprintf(stderr, "Usage: %s <ip> <port>\n", argv[0]);
        return 1;
    }
    int sock = connect_to_server(argv[1], atoi(argv[2]));
    if (sock < 0) {
        perror("Connection failed");
        return 1;
    }
    monitor_io(sock);
    close(sock);
    return 0;
}

2. libevent を利用した実装

libevent を使用すると、epoll の詳細な制御をライブラリに委ねることができます。イベントベースのプログラミングモデルにより、コールバック関数を登録するだけで非同期処理が可能になります。

サーバー側 (server_libevent.c)

event_base を生成し、リスニングソケットおよび接続済みソケットに対してイベントを割り当てます。

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <event2/event.h>
#include <event2/listener.h>
#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <arpa/inet.h>

void read_callback(struct bufferevent *bev, void *ctx) {
    struct evbuffer *input = bufferevent_get_input(bev);
    char buf[1024];
    size_t n = evbuffer_remove(input, buf, sizeof(buf) - 1);
    
    if (n > 0) {
        buf[n] = '\0';
        printf("Data: %s", buf);
        bufferevent_write(bev, "OK\n", 3);
    }
}

void event_callback(struct bufferevent *bev, short events, void *ctx) {
    if (events & BEV_EVENT_EOF) {
        printf("Connection closed\n");
    } else if (events & BEV_EVENT_ERROR) {
        perror("Got an error on the connection");
    }
    bufferevent_free(bev);
}

void accept_callback(struct evconnlistener *listener, evutil_socket_t fd, 
                     struct sockaddr *sa, int socklen, void *ctx) {
    struct event_base *base = evconnlistener_get_base(listener);
    struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
    
    bufferevent_setcb(bev, read_callback, NULL, event_callback, NULL);
    bufferevent_enable(bev, EV_READ|EV_WRITE);
}

int main() {
    struct event_base *base = event_base_new();
    struct sockaddr_in sin;
    
    memset(&sin, 0, sizeof(sin));
    sin.sin_family = AF_INET;
    sin.sin_port = htons(9000);
    sin.sin_addr.s_addr = htonl(0);

    struct evconnlistener *listener = evconnlistener_new_bind(base, accept_callback, NULL,
        LEV_OPT_REUSEABLE|LEV_OPT_CLOSE_ON_FREE, -1, (struct sockaddr*)&sin, sizeof(sin));
    
    if (!listener) {
        perror("Couldn't create listener");
        return 1;
    }

    event_base_dispatch(base);
    evconnlistener_free(listener);
    event_base_free(base);
    return 0;
}

クライアント側 (client_libevent.c)

クライアント側でも bufferevent を利用することで、ソケットと標準入力のイベント処理を統一的に扱えます。

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <event2/event.h>
#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <arpa/inet.h>

void server_read_cb(struct bufferevent *bev, void *ctx) {
    struct evbuffer *input = bufferevent_get_input(bev);
    char buf[1024];
    size_t n = evbuffer_remove(input, buf, sizeof(buf) - 1);
    if (n > 0) {
        buf[n] = '\0';
        printf("%s", buf);
        fflush(stdout);
    }
}

void server_event_cb(struct bufferevent *bev, short events, void *ctx) {
    if (events & BEV_EVENT_EOF) {
        printf("\nServer closed connection\n");
        exit(0);
    }
    bufferevent_free(bev);
}

void stdin_read_cb(evutil_socket_t fd, short what, void *arg) {
    struct bufferevent *bev = (struct bufferevent *)arg;
    char buf[1024];
    ssize_t n = read(fd, buf, sizeof(buf));
    if (n > 0) {
        bufferevent_write(bev, buf, n);
    } else {
        exit(0);
    }
}

int main(int argc, char **argv) {
    if (argc != 3) {
        fprintf(stderr, "Usage: %s <ip> <port>\n", argv[0]);
        return 1;
    }

    struct event_base *base = event_base_new();
    struct sockaddr_in sin;
    
    sin.sin_family = AF_INET;
    sin.sin_port = htons(atoi(argv[2]));
    inet_pton(AF_INET, argv[1], &sin.sin_addr);

    evutil_socket_t sockfd = socket(AF_INET, SOCK_STREAM, 0);
    evutil_make_socket_nonblocking(sockfd);

    struct bufferevent *bev = bufferevent_socket_new(base, sockfd, BEV_OPT_CLOSE_ON_FREE);
    bufferevent_setcb(bev, server_read_cb, NULL, server_event_cb, NULL);
    bufferevent_enable(bev, EV_READ|EV_WRITE);

    if (bufferevent_socket_connect(bev, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
        perror("Connection failed");
        bufferevent_free(bev);
        return 1;
    }

    struct event *ev_stdin = event_new(base, STDIN_FILENO, EV_READ|EV_PERSIST, stdin_read_cb, bev);
    event_add(ev_stdin, NULL);

    event_base_dispatch(base);
    return 0;
}