neoagent1人読書会 その3


前回はna_event_loopを読んだ。その中でna_front_server_callbackを下のようにイベントループに登録していた。

```c
    env->fs_watcher.data = env;
    ev_io_init(&env->fs_watcher, na_front_server_callback, env->fsfd, EV_READ);
    ev_io_start(EV_A_ &env->fs_watcher);
    ev_loop(EV_A_ 0);
```

evn->fs_watcherはev_io型で、コールバック関数の引数として渡される。このdataメンバーには(void *)を格納できて、コールバック関数に好きなものを渡すことができる。

今回はこのna_front_server_callbackを読んでいく。

```c
void na_front_server_callback (EV_P_ struct ev_io *w, int revents)
{
    int fsfd, cfd, tsfd, cur_pool, cur_cli;
    int th_ret;
    na_env_t *env;
    na_client_t *client;
    na_connpool_t *connpool;

    th_ret   = 0;
    fsfd     = w->fd;
    env      = (na_env_t *)w->data;
    cfd      = -1;
    tsfd     = -1;
    cur_pool = -1;
    cur_cli  = -1;
```

libevのコールバック関数の型。ev_io wというのはさっき見たfs_watcherが入っているので、w->dataをna_env_tにキャストしなおしてenvを得ている。

```c
..

    pthread_rwlock_rdlock(&env->lock_refused);
    if (env->is_refused_accept) {
        pthread_rwlock_unlock(&env->lock_refused);
        return;
    }
    pthread_rwlock_unlock(&env->lock_refused);
    
    if (env->error_count_max > 0 && (env->error_count > env->error_count_max)) {
        env->error_count = 0;
        return;
    }
    
    pthread_mutex_lock(&env->lock_current_conn);
    if (env->current_conn >= env->conn_max) {
        pthread_mutex_unlock(&env->lock_current_conn);
        return;
    }
    pthread_mutex_unlock(&env->lock_current_conn);
```

env（≒グローバル）の状態によってこのコールバックは終了。ウェブアプリケーション側からはKVSに接続できなかったという状態に見えるのかな…？（よくわからない）

```c
    connpool = na_connpool_select(env);
```

na_connpool_selectはconnpool.cにある。アクティブのKVS用のコネクションプールかスタンバイのKVS用のコネクションプールを選択するらしい。基本はアクティブで、env->is_refused_activeだったらスタンバイということになっている。

```c
    if (env->is_connpool_only) {
        pthread_mutex_lock(&env->lock_current_conn);
        if (env->current_conn >= connpool->max) {
            pthread_mutex_unlock(&env->lock_current_conn);
            return;
        }
        pthread_mutex_unlock(&env->lock_current_conn);
        if (!na_connpool_assign(env, &cur_pool, &tsfd)) {
            na_error_count_up(env);
            NA_STDERR("failed assign connection from connpool.");
            return;
        }
    } else {
```

コネクションプールのみの設定になっている場合、コネクション数の上限を超えてたりna_connpool_assign(env, &cur_pool, &tsfd)できなければreturn。

na_connpool_assignはコネクションプールから適当に一つ選んで、番号をcur_poolに、ソケットをtsfdに書きだして返すらしい。コネクションプールがいっぱいならfalseを返す。

次のelseはコネクションプールのみじゃない設定のとき。その場合、コネクションプールがいっぱになったら新たにKVSへのコネクションを開くことになるらしい。

```c
    if ((cfd = na_server_accept(fsfd)) < 0) {
        if (cur_pool == -1) {
            close(tsfd);
        } else {
            pthread_mutex_lock(&env->lock_connpool);
            connpool->mark[cur_pool] = 0;
            pthread_mutex_unlock(&env->lock_connpool);
        }
        NA_STDERR_MESSAGE(NA_ERROR_INVALID_FD);
        return;
    }
```

ここではfsfdをacceptしてcfdを作っている。

最初はこれがよく分からなかったのだけど、socketプログラミングにおいては常識の概念だそうで、サーバー側のソケットというのはsocket()で生成したソケットをbind()でポートに登録して、listen()で待ち受け状態にしてaccept()を呼ぶことで接続を確立するというフローになるらしい。ただし、このソケットに直接read()/write()することはできない。実はaccept()は別のソケットを生成して返してくるので、accept()によって作られたソケットに対して読み書きするらしい。

つまり、fsfdはウェブアプリケーションからのリクエスト待ち専用のソケット、cfdは読み書き専用のソケットということになる。

```c
    na_set_nonblock(cfd);
```

これはそのソケットに対してNONBLOCKフラグをつけている。これを立てると、readしようとしたときにまだ読むべきデータが届いてない場合に（待ち続けずに）失敗してくれるらしい。入力待ちになるのをepollとかで監視することで、IO待ちの発生しないプログラムになる。そのへんの面倒なことをうまくやってくれるのがlibevなのだと思う（自信ない）。

```c
    cur_cli = na_client_assign(env);

    if (cur_cli >= 0) {
        client = &ClientPool[cur_cli];
        if (client->tsfd > 0) {
            close(client->tsfd);
        }
    } else {
```

na_client_assignはクライアントプールから空いてるクライアントを適当に一つ選ぶ関数。空きがあればそのクライアントを使う。

neoagentでは、①ウェブアプリケーションからのリクエストを読み込んで、②KVSに書き出して、③KVSからのレスポンスをまた読み込んで、④またウェブアプリケーションに返すというのを行うのに「クライアント」（na_client_t）というものを割り当てている。クライアントはウェブアプリケーションからのリクエストそれぞれに対して一つずつあり、現在自分が①〜④のどの状態にあるかを管理するステートマシンとなる。

```c
        client = (na_client_t *)malloc(sizeof(na_client_t));
        if (client == NULL) {
            close(cfd);
            if (cur_pool == -1) {
                close(tsfd);
            } else {
                pthread_mutex_lock(&env->lock_connpool);
                connpool->mark[cur_pool] = 0;
                pthread_mutex_unlock(&env->lock_connpool);
            }
            na_error_count_up(env);
            NA_STDERR_MESSAGE(NA_ERROR_OUTOF_MEMORY);
            return;
        }
        memset(client, 0, sizeof(*client));
        client->crbuf = (char *)malloc(env->request_bufsize + 1);
        client->srbuf = (char *)malloc(env->response_bufsize + 1);
        if (client->crbuf == NULL ||
            client->srbuf == NULL) {
            NA_FREE(client->crbuf);
            NA_FREE(client->srbuf);
            NA_FREE(client);
            close(cfd);
            if (cur_pool == -1) {
                close(tsfd);
            } else {
                pthread_mutex_lock(&env->lock_connpool);
                connpool->mark[cur_pool] = 0;
                pthread_mutex_unlock(&env->lock_connpool);
            }
            na_error_count_up(env);
            NA_STDERR_MESSAGE(NA_ERROR_OUTOF_MEMORY);
            return;
        }
```

クライアントプールに空いてるのがなければmallocする。

```c
    client->cfd               = cfd;
    client->tsfd              = tsfd;
    client->env               = env;
    client->c_watcher.data    = client;
    client->ts_watcher.data   = client;
    pthread_rwlock_rdlock(&env->lock_refused);
    client->is_refused_active = env->is_refused_active;
    pthread_rwlock_unlock(&env->lock_refused);
    client->is_use_connpool   = cur_pool != -1 ? true : false;
    client->is_use_client_pool = cur_cli != -1 ? true : false;
    client->cur_pool          = cur_pool;
    client->crbufsize         = 0;
    client->cwbufsize         = 0;
    client->srbufsize         = 0;
    client->swbufsize         = 0;
    client->request_bufsize   = env->request_bufsize;
    client->response_bufsize  = env->response_bufsize;
    client->event_state       = NA_EVENT_STATE_CLIENT_READ;
    client->req_cnt           = 0;
    client->res_cnt           = 0;
    client->loop_cnt          = 0;
    client->cmd               = NA_MEMPROTO_CMD_NOT_DETECTED;
    client->connpool          = na_connpool_select(env);

..
```

クライアントはcfd（ウェブアプリケーションと通信するソケット）とtsfd（KVSと通信するソケット）を持ち、それぞれのソケットをイベントループで監視するためのc_watcherとts_watcherを持つ。

いよいよ今回の最後。

```c
    if (!na_is_worker_busy(env)) {
        if (!na_event_queue_push(EventQueue, client)) {
            na_error_count_up(env);
            NA_STDERR("Too Many Connections!");
            ev_io_init(&client->c_watcher,  na_client_callback,        client->cfd,  EV_READ);
            ev_io_init(&client->ts_watcher, na_target_server_callback, client->tsfd, EV_NONE);
            ev_io_start(EV_A_ &client->c_watcher);
        }
    } else {
        ev_io_init(&client->c_watcher,  na_client_callback,        client->cfd,  EV_READ);
        ev_io_init(&client->ts_watcher, na_target_server_callback, client->tsfd, EV_NONE);
        ev_io_start(EV_A_ &client->c_watcher);
    }
}
```

まずna_is_worker_busyというのがある。workerは前回出てきた、na_event_observerのスレッドのこと。workerスレッドがbusyならメインスレッドでウェブアプリケーションからリクエストを読み込むイベントループ（c_watcher）を開始する。

workerがbusyじゃなければクライアントをイベントキューに追加。

キューがいっぱいならメインスレッドでイベントループをまわす。

長かった。まとめると、na_front_server_callbackはウェブアプリケーションからの接続があると呼ばれるコールバックで、クライアントというものを初期化して、KVSに接続して（あるいはコネクションプールから使って）、イベントキューに追加していた。

次回はna_event_observerを読む。