neoagent1人読書会 その1

neoagentはpixivが開発しているオープンソースのmemcachedプロキシで、類似ソフトにはmemagentやtwemproxyがある（らしい）。

簡単に言うと、ウェブアプリケーションからunix domain socketで接続を受け、コネクションプーリングしつつmemcached等にクエリを流すソフト。

libev, pthread, socket等の技術が出てくる。

これを書いてる人は、C言語は読めるけど積極的に使いたくはないレベル。また、libevはnode.jsのソースを読んだりして知ってるけどpthreadやsocketはほとんど知らないレベル。

v0.5.3を読んでいく。

ソースは全部neoagentディレクトリ以下にある。

まずはneoagent.cのmain関数から。

int main (int argc, char *argv[])
{
    pthread_t           th[NA_ENV_MAX];
    na_env_t           *env[NA_ENV_MAX];
    mpool_t            *env_pool;
    int                 c;
    int                 env_cnt          = 0;
    bool                is_daemon        = false;
    struct json_object *conf_obj         = NULL;
    struct json_object *environments_obj = NULL;

envというのを何個か持つらしい。開発者の方曰く、envは1つである前提で読んでいいらしい。

    while (-1 != (c = getopt(argc, argv,
           "f:" /* configuration file with JSON */
           "t:" /* check configuration file */
           "d"  /* go to background */
           "v"  /* show version and information */
           "h"  /* show help */
    )))
    {
        switch (c) {
        case 'd':
            is_daemon = true;
            break;
        case 'f':
            conf_obj         = na_get_conf(optarg);
            environments_obj = na_get_environments(conf_obj, &env_cnt);
            break;

..

    env_pool = mpool_create(0);
    if (env_cnt == 0) {
        env_cnt = 1;
        env[0]  = na_env_add(&env_pool);
        na_env_setup_default(env[0], 0);
    } else {
        for (int i=0;i<env_cnt;++i) {
            env[i] = na_env_add(&env_pool);
            na_env_setup_default(env[i], i);
            na_conf_env_init(environments_obj, env[i], i);
        }
    }

コマンドラインオプションで与えたJSONからconfigを読み込んでenv[0]に入れるらしい。（env_cntは1の場合のみを考える）

    for (int i=0;i<env_cnt;++i) {
        env[i]->current_conn      = 0;
        env[i]->is_refused_active = false;
        env[i]->is_refused_accept = false;
        env[i]->is_worker_busy    = calloc(sizeof(bool), env[i]->worker_max);
        for (int j=0;j<env[i]->worker_max;++j) {
            env[i]->is_worker_busy[j] = false;
        }
        env[i]->error_count       = 0;
        env[i]->current_conn_max  = 0;
        pthread_mutex_init(&env[i]->lock_connpool, NULL);
        pthread_mutex_init(&env[i]->lock_current_conn, NULL);
        pthread_mutex_init(&env[i]->lock_tid, NULL);
        pthread_mutex_init(&env[i]->lock_loop, NULL);
        pthread_rwlock_init(&env[i]->lock_refused, NULL);
        env[i]->lock_worker_busy  = calloc(sizeof(pthread_rwlock_t), env[i]->worker_max);
        for (int j=0;j<env[i]->worker_max;++j) {
            pthread_rwlock_init(&env[i]->lock_worker_busy[j], NULL);
        }
        na_connpool_create(&env[i]->connpool_active, env[i]->connpool_max);
        if (env[i]->is_use_backup) {
            na_connpool_create(&env[i]->connpool_backup, env[i]->connpool_max);
        }
    }

    for (int i=0;i<env_cnt;++i) {
        pthread_create(&th[i], NULL, na_event_loop, env[i]);
    }

envを初期化して、最後にpthread_createしている。

threadに渡すのはna_event_loop関数と、その引数env[i]。

    // monitoring signal
    while (true) {

        if (SigExit == 1) {
            break;
        }

        sleep(1);
    }

あとは殺されるまでひたすら無限ループ。

今回はここまで。