I/Oの多重化(epoll API)を使いやすくしたい

はじめに

epoll APIは有能だなあと思う今日この頃ですが、epoll_wait()してイベントを取り出す処理をべた書きするとコードが汚くなりがちです。少し前の記事ですが、epoll APIを使ったチャットサーバでは、epoll_wait()後のfor(){}が長くなり、読みにくいものになっていました。そこでイベント監視を少し一般化して裏でepollを呼ぶような仕組みを作りたくなりました。話がややこしいので結論だけ書いておくと、

epoll_data_t dataのユーザ定義のポインタに関数ポインタが混じった構造体を代入すると簡潔にアプリケーション側が書けそう
インタレストリストへ追加/削除するような基本的な関数を作ってアプリケーション側から呼ぶと良さそう
だねっていう話です。

後述するこれらの良くないであろう点でも書いていますが、あまりいいものではないです。この記事内で使うコードの一式をGitHubに置いておきます。ヘッダファイルはこの記事内ですべてを載せてるわけではなので使ってみたい場合は以下のリポジトリからどうぞ。

いい感じにするための方針

方向性

いい感じにするには、struct epoll_eventのメンバであるepoll_data_t dataカギになりそうです。epoll_data_tは共用体なので実際に(同時に)使えるメンバは1つだけです。前回のチャットサーバと同様に、void *ptrを使うことを考えていきます。void *ですので任意の型で代入することができます。

typedef union epoll_data {
    void *ptr; // ユーザ定義のデータへのポインタ
    int fd; // ファイルディスクリプタ
    uint32_t u32; // 32ビット整数
    uint64_t u64; // 64ビット整数
} epoll_data_t;

前回のチャットサーバで、for(){}が長くなっている原因はIO可能になって処理する内容を関数として作られていないからというのがありますが、何よりもそれらの処理が連結リストのポインタの操作が混じっていることが原因だと思います。そのため、ポインタ操作など面倒くさいものはできるだけ裏でやるようにすると良さそうです。

void*ptrに入れる構造体

void *ptrに代入するものを以下のstruct io_eventのようにしました。

/**
 * epollのstruct epoll_eventのユーザ定義のフィールドで使いたい構造体
 */
struct io_event {
    //! 監視対象のファイルディスクリプタ
    int fd;
    //! そのイベントがIO可能になった時に呼び出される関数(ポインタ)
    void (*handler)(struct io_event*);
    //! 呼び出される関数内で使いたい変数など(任意)
    void* arg;
    //! 前のio_event(連結リスト)
    struct io_event* prev;
    //! 後のio_event(連結リスト)
    struct io_event* next;
    //! 最後に発生したイベント日時
    struct tm* timestamp;
    //! 発生したイベントの種類
    observe_type type;
};

int fd
struct epoll_eventのメンバであるepoll_data_t dataは共用体なので、void *ptrを使うとファイルディスクリプタを代入する変数int fdは使えなくなり、そのままだと何かと不便なので以下のstruct io_eventにファイルディスクリプタの変数を作りました。
void (*handler)(struct io_event*)
IO可能となった時に呼び出される関数を関数ポインタvoid (*handler)(struct io_event*);としました。その引数をstruct io_event*とし、そのIO可能となったファイルディスクリプタのイベント情報を渡せるようにしました。
void* arg
void (*handler)(struct io_event*);内で使いたい変数などを入れておくための変数。例えば、クライアント-サーバモデルを考えたときにファイルディスクリプタにユーザ名結び付けた場合は、この変数にユーザ名を入れておくことでユーザ名の登録後の発言で、発言内容と発言元のユーザ名が取れるようになります。
struct io_event* prev、struct io_event* next;
連結(双方向)リストのためのポインタ、説明不要
struct tm* timestamp
IOが可能となった時点のタイムスタンプ
observe_type type
監視するイベントの種類(以下の列挙体)

/**
 * @enum observe_type
 * 監視するイベントの種類が定義された列挙体
 */
typedef enum {
    //! EPOLLINに相当
    OBS_IN = 1u,
    //! EPOLLOUTに相当
    OBS_OUT = 4u,
    //! EPOLLONESHOTに相当
    OBS_ONESHOT = 1u << 30,
} observe_type;

/**
 * event_io全体で使われる変数一式が定義された構造体
 */
struct event_config {
    //! epollのファイルディスクリプタ
    int epfd;
    //! struct io_eventの先頭ポインタ
    struct io_event* head;
    //! struct io_eventの末尾ポインタ
    struct io_event* tail;
};

これらを使って基本的なイベント監視の登録/解除の操作を作っていきます。

基本的な操作

監視をするための初期化

/**
 * イベント監視の初期化処理をする関数
 * @return strcut event_config*
 */
struct event_config* init_event()
{
    struct event_config* event_config = malloc(sizeof(struct event_config));
    if (event_config == NULL) {
        perror("malloc()");
        exit(EXIT_FAILURE);
    }
    // epollのインスタンスを作成
    event_config->epfd = epoll_create(MAX_EPOLL);
    if (event_config->epfd == -1) {
        perror("epoll_create()");
        exit(EXIT_FAILURE);
    }
    event_config->head = NULL;
    event_config->tail = NULL;
    return event_config;
}

監視対象の追加

/**
 * イベント監視を新たに追加する関数
 * @param struct event_config* event_config
 * @param void (*handler)(struct io_event*) イベント発生時に呼び出すための関数ポインタ
 * @param void* arg handlerを呼び出すときの引数
 * @param int fd 監視対象のファイルディスクリプタ
 * @return struct io_event*
 */
struct io_event* add_event(struct event_config* event_config, void (*handler)(struct io_event*), void* arg, int fd, observe_type type)
{
    if (event_config == NULL) {
        perror("NULL");
        exit(EXIT_FAILURE);
    }
    struct io_event* io = malloc(sizeof(struct io_event));
    if (io == NULL) {
        perror("malloc()");
        exit(EXIT_FAILURE);
    }
    io->handler = handler;
    io->arg = arg;
    io->fd = fd;
    io->next = NULL;
    io->prev = NULL;
    io->timestamp = NULL;
    // まだ監視対象がないとき(初回)
    if (event_config->head == NULL) {
        event_config->head = io;
        event_config->tail = io;
    } else {
        // 監視対象が少なくとも1つあるときは末尾につなげていく
        event_config->tail->next = io;
        io->prev = event_config->tail;
        event_config->tail = io;
    }
    struct epoll_event event;
    event.events = type;
    event.data.ptr = io;
    // インタレストリスト(監視対象のリスト)に追加する
    // fdが既にインタレストリストに存在する場合はEEXIST
    if (epoll_ctl(event_config->epfd, EPOLL_CTL_ADD, fd, &event) == -1) {
        perror("epoll_ctl()");
        exit(EXIT_FAILURE);
    }
    return io;
}

特定の監視対象の削除

/**
 * イベント監視を外す関数
 * @param struct event_config* event_config
 * @param struct io_event* io 監視対象を外したいイベント
 * @return void
 */
void delete_event(struct event_config* event_config, struct io_event* io)
{
    if (event_config == NULL) {
        perror("NULL");
        exit(EXIT_FAILURE);
    }
    // 監視対象から外す
    // fdがインタレストリストに存在しない場合は、ENOENT(no entry)
    if (epoll_ctl(event_config->epfd, EPOLL_CTL_DEL, io->fd,  NULL) == -1) {
        perror("epoll_ctl()");
        exit(EXIT_FAILURE);
    }
    if (io->prev != NULL) {
        // io_eventがリストが途中の要素なら
        // 前の要素のnextと次の要素をつなげる
        io->prev->next = io->next;
    } else {
        // io_eventがリストの先頭の要素なら
        // 先頭の位置を進める
        event_config->head = io->next;
    }
    if (io->next != NULL) {
        // io_eventがリストの途中の要素なら
        // 次の要素のprevと前の要素をつなげる
        io->next->prev = io->prev;
    } else {
        // io_eventがリストの末尾の要素なら
        // 末尾の位置を戻す
        event_config->tail = io->prev;
    }
    free(io);
    io = NULL;
}

全ての監視対象の削除

/**
 * add_eventで追加した全てのイベントを削除する関数
 * @param struct event_config* event_config
 * @return void
 */
void erase_events(struct event_config* event_config)
{
    if (event_config == NULL) {
        perror("NULL");
        exit(EXIT_FAILURE);
    }
    struct io_event* head;
    while ((head = event_config->head) != NULL) {
        // 監視対象から外す
        // fdがインタレストリストに存在しない場合は、ENOENT(no entry)
        if (epoll_ctl(event_config->epfd, EPOLL_CTL_DEL, head->fd, NULL) == -1) {
            perror("epoll_ctl()");
            exit(EXIT_FAILURE);
        }
        free(head);
        head = NULL;
        event_config->head = event_config->head->next;
    }
}

IO可能になるまで待機

/**
 * イベントを走らせる関数
 * @param struct event_config* event_config
 * @param int timout IO可否に関係なく抜けるタイムアウト値(ミリ秒)
 * @return void
 */
void run_event(struct event_config* event_config, int timeout)
{
    if (event_config == NULL) {
        perror("NULL");
        exit(EXIT_FAILURE);
    }
    struct epoll_event events[MAX_EPOLL];
    // 監視対象がIO可能になるか、timoutまでブロックされる
    int event_readable = epoll_wait(event_config->epfd, events, MAX_EPOLL, timeout);
    if (event_readable == -1) {
        perror("epoll_wait()");
        exit(EXIT_FAILURE);
    }
    // IO可能になった時点のタイムスタンプ
    time_t t = time(NULL);
    if (t == -1) {
        perror("time()");
        exit(EXIT_FAILURE);
    }
    struct tm* time = localtime(&t);
    for (int i = 0; i < event_readable; i++) {
        struct io_event* io = events[i].data.ptr;
        io->timestamp = time;
        io->type = events[i].events;
        io->handler(io);
    }
}

サンプル

標準入力だけ監視してみる

まずは、動作を知るために簡単に標準入力だけをインタレストリストに追加してIO可能になるまで待ってみましょう。main関数内でイベント周りの初期化処理を行い、標準入力を監視対象として追加します。あとはrun_event()の第2引数(タイムアウト)に-1を指定して実行してやると入力があるまでブロックします。

/**
 * @file 01.c
 * @brief 単純な例
 * @author E.Okazaki
 */
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include "event_io.h"

void tirm_nl(char* str)
{
    char* p;
    p = strchr(str, '\n');
    if (p != NULL) {
        *p = '\0';
    }
}

// コールバック関数とする場合は必ずstruct io_eventを受け取る
void print(struct io_event* io)
{
    char buf[256];
    read(io->fd, buf, sizeof(buf));
    tirm_nl(buf);
    printf("FD: %d\n", io->fd);
    printf("arg: %s\n", (char*)io->arg);
    printf("event type: %d\n", io->type);
    printf("timestamp: %s", asctime(io->timestamp));
    printf("input: %s\n", buf);
    printf("---------------------------------------\n");
}

int main()
{
    struct event_config* config;
    config = init_event();
    add_event(config, print, "標準入力だよ", STDIN_FILENO, OBS_IN);
    while (1) {
        run_event(config, -1);
    }
}

マルチクライアント対応のエコーサーバを作ってみる

動作自体は前回のチャットサーバと同じです(受信したものを加工する処理を抜いたという点を除けば)。複数のクライアントから接続を受け付けられるし、あるクライアントからデータを受信したら、それを接続済みの全クライアントに対して送信するプログラムです。イベント周りの処理はサーバとクライアントのファイルディスクリプタを追加することです。サーバの方はinit_server()の最終行で、クライアントの方はaccept()_内の最終行で追加しています。

/**
 * @file 02.c
 * @brief サーバプログラムの例
 * @author E.Okazaki
 */
#include <stdio.h>
#include <unistd.h>
#include <signal.h>
#include <string.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include "event_io.h"

/// 送受信できるバイト数
#define BUF_LEN 1024
/// サーバのポート番号
#define PORT 8080

/**
 * サーバに関する情報を持つ構造体
 */
struct server {
    //! サーバのアドレス情報
    struct sockaddr_in address;
    //! サーバのファイルディスクリプタ
    int fd;
    //! 送受信に使う変数
    char buf[BUF_LEN];
    //! イベントに関する情報を持つ構造体
    struct event_config* event;
};

void init_server(struct server* server);
void write_(struct io_event* head, char* msg);
void read_(struct io_event* io);
void accept_(struct io_event* io);

/**
 * サーバを立ち上げる関数
 * 立ち上げが完了したらそのファイルディスクリプタをイベント監視に追加する
 * @param struct server* server
 * @return void
 */
void init_server(struct server* server)
{
    int opt = 1;
    // ソケット(エンドポイント)を作成する
    if ((server->fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
        perror("socket()");
        exit(EXIT_FAILURE);
    }
    // オプションを付加する
    if (setsockopt(server->fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT,
                   &opt, sizeof(opt))) {
        perror("setsockopt()");
        exit(EXIT_FAILURE);
    }
    // アドレスファミリの設定
    server->address.sin_family = AF_INET;
    // IPv4のワイルドカード
    server->address.sin_addr.s_addr = INADDR_ANY;
    // ポート設定
    server->address.sin_port = htons(PORT);
    // ソケットへアドレスを付加する
    if (bind(server->fd, (struct sockaddr*)&server->address, sizeof(server->address)) < 0) {
        perror("bind()");
        exit(EXIT_FAILURE);
    }
    // 待ち行列を5個にしてserver_fdをパッシブ(リッスン)ソケットとしてマークする
    if (listen(server->fd, 5) < 0) {
        perror("listen()");
        exit(EXIT_FAILURE);
    }
    // サーバのファイルディスクリプタを
    // イベントの監視対象として追加する。
    // IO可能になった(新規接続が来た)時にaccept_()を実行するよう登録する。
    add_event(server->event, accept_, server, server->fd, OBS_IN);
}

/**
 * クライアントへ文字列を送信する関数
 * @param struct io_event* head クライアントのイベント情報の先頭ポインタ
 * @param char* msg 送信したい文字列
 * @return void
 * @details headがNULLになるまで再帰的にポインタを進め、クライアントへ書き込む。
 */
void write_(struct io_event* head, char* msg)
{

    if (head == NULL) {
        return;
    }
    write(head->fd, msg, strlen(msg));
    write_(head->next, msg);
}

/**
 * クライアントのイベント発生時の処理をする関数
 * @param struct io_event* io イベントが発生したクライアントのイベント情報
 * @return void
 */
void read_(struct io_event* io)
{
    struct server* server = (struct server*)io->arg;

    // 読み取り時に何らかのエラーが発生したら切断処理をする。
    if (read(io->fd, server->buf, sizeof(server->buf)) <= 0) {
        // イベントの監視を外す(delete_eventでメモリの解放も行っている)
        delete_event(server->event, io);
        return;
    }
    // 接続済みの全クライアントに文字列を送信する
    write_(server->event->head->next, server->buf);
    printf("%s\n", server->buf);
    // 0埋めしておく
    memset(server->buf, 0, sizeof(server->buf));
}

/**
 * 接続要求を受け入れる関数
 * @param struct io_event* io サーバのファイルディスクリプタのイベント情報
 * @return void
 * @details クライアントからの接続要求を受け入れ、そのクライアントのファイルディスクリプタを監視対象に追加する。
 * IO可能になった(メッセージが来た)ときにread_()を実行するように登録する。
 */
void accept_(struct io_event* io)
{
    int new_socket;
    struct server* server = (struct server*)io->arg;
    struct sockaddr_in address =  server->address;
    int addrlen = sizeof(address);
    if ((new_socket = accept(io->fd, (struct sockaddr*)&address, (socklen_t*)&addrlen)) == -1) {
        perror("accept()");
        return;
    }
    add_event(server->event, read_, server, new_socket, OBS_IN);
}

int main()
{
    // サーバに関する情報を保持する構造体
    struct server server;
    // 監視するイベント処理をするための初期化処理
    server.event = init_event();
    // サーバを立ち上げる
    init_server(&server);
    printf("サーバを立ち上げました。\n");
    while (1) {
        // イベント監視処理を開始する
        run_event(server.event, -1);
    }
}

これらの良くないであろう点

「これら」というか、これまで話に触れなかったstruct event_configについてですが、この構造体は主に裏側(イベント監視追加/削除などが書かれたevent_io.c)で使うことを想定した構造体です。しかし、上記のサーバのサンプルのようにこの構造体のheadやtailにアクセスすることができます。裏ではこれら(特にtail)を操作して新たな要素を追加/削除する処理をしているので、アプリケーション側で参照するだけなら問題ありませんが、せっかく裏で管理しているのに外から下手に弄るとリストの構造が崩壊する可能性があります。そういう意味で本来であれば、隠蔽したりイテレータみたいなものを実装して、外側からは弄れないようにするべきかなと思います。

はじめに#

いい感じにするための方針#

方向性#

void*ptrに入れる構造体#

基本的な操作#

監視をするための初期化#

監視対象の追加#

特定の監視対象の削除#

全ての監視対象の削除#

IO可能になるまで待機#

サンプル#

標準入力だけ監視してみる#

マルチクライアント対応のエコーサーバを作ってみる#

これらの良くないであろう点#