I/O多重化でチャットサーバを作る

チャットサーバの簡単な説明

1プロセス1スレッドで動作するチャットサーバを考えていきます。
echoサーバとほぼ同じです。
複数のクライアントが接続できるサーバを考えます。
クライアントからの1回目のwrite()でユーザ名の登録をします。
ユーザ名の登録以外(2回目以降)のwrite()はサーバはそれを受け取った後に、メッセージを他の全クライアントに対して送信します。
サーバはメッセージを受け取った後に以下のフォーマットに従って編集します。

[メッセージの通し番号 発言元のユーザ名 日時]
    メッセージ

プログラムの実行例(というか最終的に出来上がったものの実行例)
※ 今回はサーバがメインです。クライアント側はtelnetをいくつか開いて、それぞれをクライアントと想定しています。
発言元はログイン時、ログアウト時は"server"とし、それ以外の発言はユーザ名を表示しています。

epoll APIについて

I/Oの多重化を扱うにはいくつかの方法があるようですが、select()やpoll()よりも扱いやすそうだったため、今回はepoll()を使用しました。 epoll APIはLinux固有の機能であるためLinux上での実行を想定しています。

int epoll_create(int size);

epollインスタンスの作成
インタレストリストを空にする

引数	説明	備考
int size	監視対象のファイルディスクリプタ数	sizeは初期化処理のヒントのためであって、sizeを超えた数のファイルディスクリプタの処理も可能(だから適当な値でいいっぽい？)

戻り値の型	説明
int	成功時にファイルディスクリプタ、エラー時に-1

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *ev);

インタレストリストの変更(追加、編集、削除)

引数	説明
int epfd	epoll_create();で作成したfd
int op	EPOLL_CTL_ADD: epfd上のインタレストリストにfdを追加する EPOLL_CTL_MOD: epfd上のインタレストリストに追加済みのfdを指定されたイベントに変更する EPOLL_CTL_DEL : epfd上のインタレストリストからfdを削除する(監視を外す)
int fd	監視対象のファイルディスクリプタ
struct epoll_event* ev	epollイベント.eventsとユーザデータ.dataが格納されたstruct epoll_event

戻り値の型	説明
int	成功時に0、エラー時に-1

int epoll_wait(int epfd, struct epoll_event *evlist, int maxevents, int timeout);

イベントを待つ

引数	説明	備考
int epfd	epoll_create();で作成したfd
struct epoll_event *evlist	I/O可能なディスクリプタを割り当てるための配列
int maxevents	1度に得られる最大の数(evlistの要素数と同じでよい)
int timeout	タイムアウト時間(ミリ秒)	0が指定されたときはブロックしない -1が指定されたときは監視対象のいずれかにイベントが発生するまでブロックする

戻り値の型	説明
int	成功時にI/O可能なファイルディスクリプタ数、タイムアウト時に0、エラー時に-1

実装でハマりかけたところ

イベントの監視をするには以下の構造体が不可欠ですが、今回は単にfdを扱うのではなくて、fdとユーザ名の両方の管理が必要です。

struct epoll_event {
    uint32_t events; // epollイベント
    epoll_data_t data; // ユーザデータ
};

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

TLPIにもdataフィールドであるepoll_dataのvoid* ptrはユーザ定義であると書いてあったので、これを使えばいける！とまではなりました。しかし、dataが共用体であることを見逃していたせいで、実装時に

struct epoll_event event;
event.events = EPOLLIN;
event.data.fd = fd;
event.data.ptr = malloc(...);
epoll_ctl(epfd, EPOLL_CTL_ADD, &event);

みたいなことをして、ずっとsegmentation faultになっていました。(これに気が付くのに時間がかかってしまった。)
正しくやるならば、例えば

struct ClientInfo{
    int fd;
    int user_id;
    char *user_name;
}

をepoll_eventのdataフィールドに追加したいときは、

struct epoll_event event;
struct ClientInfo client_info;
event.events = EPOLLIN;
event.data.ptr = malloc(sizeof(struct ClientInfo));
((struct ClientInfo*)event.data.ptr)->fd = ... ;
((struct ClientInfo*)event.data.ptr)->user_id = ... ;
((struct ClientInfo*)event.data.ptr)->user_name = ... ;
epoll_ctl(epfd, EPOLL_CTL_ADD, &event);

のように、dataがメンバを2つ以上持たないようにする必要があります。

チャットサーバの実装

プリプロセッサ(記号定数)

名前	値	説明
PORT	8080	TCPポートは8080としています。
BUF_LEN	1024	各fdからの読み込み、書き込みの最大バイト数を1024としています。
MAX_EPOLL	80	一度のepoll_wait()で得られる最大イベント数を80までとしています。

#define PORT 8080
#define BUF_LEN 1024
#define MAX_EPOLL 80

FDとユーザ名を管理する構造体

今回はサーバ側のソケットfdとクライアントからの接続管理することに加え、クライアントからはユーザ名を登録できるようにしたいので、epoll_eventのdataフィールド(共用体union epoll_data)のvoid *ptrを使ってあげる必要があります。このvoid *ptrに構造体のアドレスを代入することでepoll_waitを抜けるとき(監視対象がイベント発生時)evlistから取得できます。つまり、evlistの要素が読み込み可能なファイルディスクリプタであることを示しています。また、今回は各fdをEPOLLINのみを指定し監視するので、書き込みはFDInfoを順に辿っていくよう単方向リストを実装します。

// epoll_eventのdata(ユーザ定義)で使う構造体
struct FDInfo {
    int fd;
    char* name; // ユーザー名(サーバソケットなら"server")

    // 受け取るイベントは「読み込み可能」な状態かだけにしたいので、
    // 書き込み時にfdを単方向リストを使ってたどって送信する
    struct FDInfo* next;
};

主に重要な関数

監視対象の追加

引数
- int epfd : epollインスタンスのfd
- struct FDInfo* fd_current_ptr : 現在のstruct FDInfoをポインタ(main関数を参照してください。)
- int target_fd : 監視をしたいfd
戻り値
- struct FDInfo* : 監視を追加したepoll_eventのdataフィールド内のptr

// 新しくインタレストリストにソケットを追加する。
// また、イベントepoll_eventのdataフィールドをユーザ定義にして、
// ユーザ名を扱えるようにする。
// ただし、初回read()時がユーザ名なので、この関数ではfdの代入のみ。
// epoll_eventのeventsはEPOLLINする
struct FDInfo* add_fd_to_interest_list(int epfd, struct FDInfo* fd_current_ptr, int target_fd)
{
    struct epoll_event event;
    // 取りたいイベントを「読み込み可能」にする。
    event.events = EPOLLIN;
    // ユーザ定義のデータ(FDInfo)へのポインタである.data.ptrへのメモリ確保、アドレス代入
    event.data.ptr = malloc(sizeof(struct FDInfo));
    if (event.data.ptr == NULL) {
        exit(EXIT_FAILURE);
    }
    // fdをユーザ定義のデータのfdフィールドに代入
    ((struct FDInfo*)event.data.ptr)->fd = target_fd;
    // nextを明示的にNULLとしておく
    ((struct FDInfo*)event.data.ptr)->next = NULL;
    // nameを明示的にNULLとしておく
    ((struct FDInfo*)event.data.ptr)->name = NULL;
    // epfdのインタレストリストへeventを登録
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, target_fd, &event) == -1) {
        exit(EXIT_FAILURE);
    }
    if (fd_current_ptr != NULL) {
        // 現在のfd_current_ptrの次に繋げる
        fd_current_ptr->next = event.data.ptr;
    }
    return event.data.ptr;
}

ユーザ名の登録

引数
- struct FDInfo* conn : ユーザ名を登録したい対象となるFDInfo構造体
- char* name : ユーザ名
戻り値
- void

// メモリ確保をして、FDInfoのnameフィールドにnameを追加する。
void add_name_to_fd_info(struct FDInfo* conn, char* name)
{
    conn->name = calloc(1, sizeof(char) * strlen(name));
    if (conn->name == NULL) {
        exit(EXIT_FAILURE);
    }
    strcpy(conn->name, name);
}

イベントの初期化

引数
- int* epfd : epollのインスタンスfd(初期化済みでなくてよい)
- int server_fd : scoket()時のfd
戻り値
- struct FDInfo* : 初期化処理をしたポインタ

// epollに関する初期化
// server_fdの監視追加
struct FDInfo* init_epoll_event(int* epfd, int server_fd)
{
    char server_name[] = "server";
    // epollのインスタンス作成
    *epfd = epoll_create(MAX_EPOLL);
    // サーバソケットをインタレストリストに追加
    struct FDInfo* fd_info_current = add_fd_to_interest_list(*epfd, NULL, server_fd);
    // fd_info_currentのnameフィールドを"server"として登録する
    add_name_to_fd_info(fd_info_current, server_name);
    return fd_info_current;
}

クライアントへの書き込み

引数
- struct FDInfo* fd_info_head : struct FDInfo*の先頭
- char* msg : 送信したいメッセージ
戻り値
- void

// fd_info_headに対して書き込みを行う
// struct FDInfoのnextを再帰的に進めていく
void write_(struct FDInfo* fd_info_head, char* msg)
{
    if (fd_info_head == NULL) {
        return;
    }
    write(fd_info_head->fd, msg, strlen(msg));
    write_(fd_info_head->next, msg);
}

切断処理

引数
- int epfd : epollインスタンスのfd
- struct FDInfo* fd_info_head : struct FDInfo*の先頭
- struct FDInfo* conn : 切断対象
戻り値
- struct FDInfo* : struct FDInfo*の末尾

// connの切断処理
struct FDInfo* discard(int epfd, struct FDInfo* fd_info_head, struct FDInfo* conn)
{
    // fdを閉じる
    close(conn->fd);
    // connが要らない、つまりポインタの繋ぎ変えをする必要がある
    struct FDInfo* h = fd_info_head;
    // 削除対象の手前のFDInfoを探す
    while (h->next != NULL && h->next->fd != conn->fd) {
        h = h->next;
    }
    // 前と後ろで付け替える(メモリ解放はfree(conn)がそれにあたる)
    h->next = h->next->next;
    // インタレストリストから削除
    epoll_ctl(epfd, EPOLL_CTL_DEL, conn->fd, NULL);
    // conn->nameを解放
    free(conn->name);
    // conn(struct FDInfo、つまりevent.data.ptr)を解放
    free(conn);
    // 一番最後の要素を探す
    while (h->next != NULL) {
        h = h->next;
    }
    return h;
}

プログラム全体

#include <unistd.h>
#include <stdio.h>
#include <sys/socket.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <string.h>
#include <sys/epoll.h>
#include <time.h>
#define PORT 8080
#define BUF_LEN 1024
// 一度のepoll_wait()で得られる最大イベント数
#define MAX_EPOLL 80

struct FDInfo;
void add_name_to_fd_info(struct FDInfo* conn, char* name);
struct FDInfo* add_fd_to_interest_list(int epfd, struct FDInfo* fd_current_ptr, int target_fd);
void edit_msg(char* msg, char* buf, char* name);
void show_name(struct FDInfo* fd_info_head);
void write_(struct FDInfo* fd_info_head, char* msg);
struct FDInfo* discard(int epfd, struct FDInfo* fd_info_head, struct FDInfo* conn);
int build_server(struct sockaddr_in* address);

// メッセージの通し番号
size_t serial_msg_num = 0;

// epoll_eventのdata(ユーザ定義)で使う構造体
struct FDInfo {
    int fd;
    char* name; // ユーザー名(サーバソケットなら"server")

    // 受け取るイベントはEPOLLINだけにしたいので、
    // 書き込み時にfdを単方向リストを使ってたどって送信する
    struct FDInfo* next;
};

// 新しくインタレストリストにソケットを追加する。
// また、イベントepoll_eventのdataフィールドをユーザ定義にして、
// ユーザ名を扱えるようにする。
// ただし、初回read()時がユーザ名なので、この関数ではfdの代入のみ。
// epoll_eventのeventsはEPOLLINする
struct FDInfo* add_fd_to_interest_list(int epfd, struct FDInfo* fd_current_ptr, int target_fd)
{
    struct epoll_event event;
    // 取りたいイベントを「読み込み可能」にする。
    event.events = EPOLLIN;
    // ユーザ定義のデータ(FDInfo)へのポインタである.data.ptrへのメモリ確保、アドレス代入
    event.data.ptr = malloc(sizeof(struct FDInfo));
    if (event.data.ptr == NULL) {
        exit(EXIT_FAILURE);
    }
    // fdをユーザ定義のデータのfdフィールドに代入
    ((struct FDInfo*)event.data.ptr)->fd = target_fd;
    // nextを明示的にNULLとしておく
    ((struct FDInfo*)event.data.ptr)->next = NULL;
    // nameを明示的にNULLとしておく
    ((struct FDInfo*)event.data.ptr)->name = NULL;
    // epfdのインタレストリストへeventを登録
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, target_fd, &event) == -1) {
        exit(EXIT_FAILURE);
    }
    if (fd_current_ptr != NULL) {
        // 現在のfd_current_ptrの次に繋げる
        fd_current_ptr->next = event.data.ptr;
    }
    return event.data.ptr;
}

// メモリ確保をして、FDInfoのnameフィールドにnameを追加する。
void add_name_to_fd_info(struct FDInfo* conn, char* name)
{
    conn->name = calloc(1, sizeof(char) * strlen(name));
    if (conn->name == NULL) {
        exit(EXIT_FAILURE);
    }
    strcpy(conn->name, name);
}

// 受信したメッセージ(buf)の編集
// 名前と時刻をメッセージに付けてmsgに出力
void edit_msg(char* msg, char* buf, char* name)
{
    char str_time[32] = {'\0'};
    time_t t = time(NULL);
    // 現在時刻の取得
    struct tm* date = localtime(&t);
    // yyyy/mm/dd hh:mm:ss
    strftime(str_time, sizeof(str_time), "%Y/%m/%d %H:%M:%S", date);
    sprintf(msg, "[%ld %s %s]\n   %s\n", ++serial_msg_num, name, str_time, buf);
}

// デバッグ用
// struct FDInfoの
// nextで繋がれているポインタを走査
void show_name(struct FDInfo* fd_info_head)
{
    if (fd_info_head == NULL) {
        return;
    }
    printf("name=%s, fd=%d\n", fd_info_head->name, fd_info_head->fd);
    show_name(fd_info_head->next);
}

// fd_info_headに対して書き込みを行う
// struct FDInfoのnextを再帰的に進めていく
void write_(struct FDInfo* fd_info_head, char* msg)
{
    if (fd_info_head == NULL) {
        return;
    }
    write(fd_info_head->fd, msg, strlen(msg));
    write_(fd_info_head->next, msg);
}

// connの切断処理
struct FDInfo* discard(int epfd, struct FDInfo* fd_info_head, struct FDInfo* conn)
{
    // fdを閉じる
    close(conn->fd);
    // connが要らない、つまりポインタの繋ぎ変えをする必要がある
    struct FDInfo* h = fd_info_head;
    // 削除対象の手前のFDInfoを探す
    while (h->next != NULL && h->next->fd != conn->fd) {
        h = h->next;
    }
    // 前と後ろで付け替える(メモリ解放はfree(conn)がそれにあたる)
    h->next = h->next->next;
    // インタレストリストから削除
    epoll_ctl(epfd, EPOLL_CTL_DEL, conn->fd, NULL);
    // conn->nameを解放
    free(conn->name);
    // conn(struct FDInfo、つまりevent.data.ptr)を解放
    free(conn);
    // 一番最後の要素を探す
    while (h->next != NULL) {
        h = h->next;
    }
    return h;
}

// CR LFを取り除く
void trim_nl(char* str)
{
    char* p;
    p = strchr(str, '\n');
    if (p != NULL) {
        *p = '\0';
    }
    p = strchr(str, '\r');
    if (p != NULL) {
        *p = '\0';
    }
}

int build_server(struct sockaddr_in* address)
{
    int server_fd, opt = 1;
    if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
        exit(EXIT_FAILURE);
    }

    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT,
                   &opt, sizeof(opt))) {
        exit(EXIT_FAILURE);
    }
    address->sin_family = AF_INET;
    address->sin_addr.s_addr = INADDR_ANY;
    address->sin_port = htons(PORT);

    if (bind(server_fd, (struct sockaddr*)address, sizeof(*address)) < 0) {
        exit(EXIT_FAILURE);
    }

    if (listen(server_fd, 5) < 0) {
        exit(EXIT_FAILURE);
    }
    return server_fd;
}

// epollに関する初期化
// server_fdの監視追加
struct FDInfo* init_epoll_event(int* epfd, int server_fd)
{
    char server_name[] = "server";
    // epollのインスタンス作成
    *epfd = epoll_create(MAX_EPOLL);
    // サーバソケットをインタレストリストに追加
    struct FDInfo* fd_info_current = add_fd_to_interest_list(*epfd, NULL, server_fd);
    // fd_info_currentのnameフィールドを"server"として登録する
    add_name_to_fd_info(fd_info_current, server_name);
    return fd_info_current;
}

int main(int argc, char* argv[])
{
    int new_socket;
    struct sockaddr_in address;
    int addrlen = sizeof(address);
    int server_fd = build_server(&address);
    printf("サーバを立ち上げました。\n");

    // epollインスタンスのfd現在なfdの数
    int epfd, event_readable;
    // 複数のイベントを取れるようにする
    struct epoll_event events[MAX_EPOLL];
    struct FDInfo* fd_info_current = init_epoll_event(&epfd, server_fd);
    // 先頭ポインタの記録用(送信時に使いたい)
    struct FDInfo* fd_info_head = fd_info_current;

    char buf[BUF_LEN] = {0};
    char msg[BUF_LEN] = {0};

    while (1) {
        // インタレストリストに追加したfdのいずれかに
        // イベントが発生するまで常にブロッキング
        // (クライアントからのconnect()もしくはsend()以外はブロッキング)
        event_readable = epoll_wait(epfd, events, MAX_EPOLL, -1);
        memset(buf, 0, sizeof(buf));
        memset(msg, 0, sizeof(msg));
        for (int i = 0; i < event_readable; i++) {
            struct FDInfo* conn = events[i].data.ptr;
            // server_fdでのイベント(クライアントからconnect())なら接続確立をする
            if (conn->fd == server_fd) {
                if ((new_socket = accept(server_fd, (struct sockaddr*)&address, (socklen_t*)&addrlen)) == -1) {
                    continue;
                }
                // インタレストリストにnew_socketを追加する
                add_fd_to_interest_list(epfd, fd_info_current, new_socket);
                // ポインタを進める
                fd_info_current = fd_info_current->next;
                // 接続確立を終えたので戻る
                continue;
            }

            // クライアントからsend()の時の処理
            // bufに読もうとしてエラーなら切断
            if (read(conn->fd, buf, sizeof(buf)) <= 0) {
                // ユーザ名が登録済みなら
                if (strlen(conn->name) > 0) {
                    sprintf(buf, "%sがログアウトしました。\n", conn->name);
                    edit_msg(msg, buf, fd_info_head->name);
                    printf("%s\n", msg);
                    write_(fd_info_head->next, msg);
                }
                // 切断処理をして、現在のポインタを更新する
                fd_info_current = discard(epfd, fd_info_head, conn);
                // 切断処理を終えたので戻る
                continue;
            }

            trim_nl(buf);

            // ユーザ名が登録されていないなら(初回のみ)
            if (conn->name == NULL) {
                // bufのメモリ領域は使いまわしたいので、
                // add_name_to_fd_info()内で新たにメモリ確保してconn->nameにコピーする。
                add_name_to_fd_info(conn, buf);
                sprintf(buf, "%sがログインしました。\n", conn->name);
                edit_msg(msg, buf, fd_info_head->name);
            } else {
                edit_msg(msg, buf, conn->name);
            }
            printf("%s\n", msg);

            // デバッグ用
            // show_name(fd_info_head->next);

            // msgをクライアントに流す
            // fd_info_headはサーバソケットで、
            // クライアントのソケットは->next以降である
            write_(fd_info_head->next, msg);
        } //  "for" 終わり
    } // "while" 終わり
}

注意点

後処理

サーバのシャットダウン時については今回考慮していません。考慮するならば、標準入力STDIN_FILENOを監視対象に追加して、“shutdown"と入力されたら

close(server_fd);
close(epfd);

をするような処理を追加するといいかもしれません。

イベント

今回はEPOLLINだけをイベント監視の対象としましたが、EPOLLOUTなどを設定したい場合もあります。 event.eventsはビットマスクですので、書き込みと読み込みを監視したいときは

event.events = EPOLLIN | EPOLLOUT;

として、イベント発生時は

if(events[i].events & EPOLLIN){
    ...
}

if(events[i].events & EPOLLOUT){
    ...
}

のように書いて処理を分けることができます。

おまけ

監視なしノンブロッキングI/Oはどうなる？

当初は以下のような単純なノンブロッキングI/Oのフラグでプログラムを作成しました。しかし、このやり方はデータの受信の有無に関係なく常にループしているので、クライアントからのリクエストがない限りEAGAINが発生し、最終的に重くなりました。

...
server_fd = socket(AF_INET, SOCK_STREAM | O_NONBLOCK, 0);
...
while(1){
   new_socket = accept(server_fd, ...);
   if(new_socket > 0){
       int flag =  fcntl(new_socket, F_GETFL, 0);
       fcntl(new_socket, F_SETFL, flag | O_NONBLOCK);
       printf("新しい接続がありました。\n");
       ...new_socketを配列に追加...
   }
   // 配列に格納済みのfdを順次読み取る
   read(...);
   // 配列に格納済みのfdに順次書き込む
   write(...);
   ...
}

参考

Manページ READ
Manページ WRITE
LINUXプログラミングインターフェース(著: Michael Kerrisk、訳者: 千住治郎)

チャットサーバの簡単な説明#

epoll APIについて#

int epoll_create(int size);#

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *ev);#

int epoll_wait(int epfd, struct epoll_event *evlist, int maxevents, int timeout);#

実装でハマりかけたところ#

チャットサーバの実装#

プリプロセッサ(記号定数)#

FDとユーザ名を管理する構造体#

主に重要な関数#

監視対象の追加#

ユーザ名の登録#

イベントの初期化#

クライアントへの書き込み#

切断処理#

プログラム全体#

注意点#

後処理#

イベント#

おまけ#

監視なしノンブロッキングI/Oはどうなる？#

参考#