<commit_before>/**
   In short, neoagent is distributed under so called "BSD license",
   
   Copyright (c) 2011-2012 <NAME> <<EMAIL>>
   All rights reserved.
   
   Redistribution and use in source and binary forms, with or without modification, 
   are permitted provided that the following conditions are met:
   
   * Redistributions of source code must retain the above copyright notice, 
   this list of conditions and the following disclaimer.
   
   * Redistributions in binary form must reproduce the above copyright notice, 
   this list of conditions and the following disclaimer in the documentation 
   and/or other materials provided with the distribution.
   
   * Neither the name of the authors nor the names of its contributors 
   may be used to endorse or promote products derived from this software 
   without specific prior written permission.
   
   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 
   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 
   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 
   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED 
   TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
   PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
   LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
   NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
   SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdbool.h>
#include <stdint.h>
#include <unistd.h>
#include <errno.h>
#include <sys/ioctl.h>
#include <pthread.h>
#include <assert.h>
#include <time.h>

#include <ev.h>

#include "event.h"
#include "env.h"
#include "socket.h"
#include "error.h"
#include "memproto.h"
#include "stat.h"
#include "util.h"

#define NA_CONNPOOL_SELECT(env) (env->is_refused_active ? &env->connpool_backup : &env->connpool_active)

// constants
static const int NA_STAT_BUF_MAX = 8192;

// external globals
volatile sig_atomic_t SigExit;

// private functions
inline static void na_event_switch (EV_P_ struct ev_io *old, ev_io *new, int fd, int revent);
inline static void na_error_count_up (na_env_t *env);

static void na_connpool_deactivate (na_connpool_t *connpool);
static bool na_connpool_assign (na_env_t *env, int *cur, int *fd);
static void na_connpool_init (na_env_t *env);
static void na_connpool_switch (na_env_t *env);
static void na_client_close (na_client_t *client, na_env_t *env);
static void na_health_check_callback (EV_P_ ev_timer *w, int revents);
static void na_stat_callback (EV_P_ struct ev_io *w, int revents);
static void na_target_server_callback (EV_P_ struct ev_io *w, int revents);
static void na_client_callback (EV_P_ struct ev_io *w, int revents);
static void na_front_server_callback (EV_P_ struct ev_io *w, int revents);
static void *na_support_loop (void *args);

inline static void na_event_switch (EV_P_ struct ev_io *old, struct ev_io *new, int fd, int revent)
{
    ev_io_stop(EV_A_ old);
    ev_io_set(new, fd, revent);
    ev_io_start(EV_A_ new);
}

inline static void na_error_count_up (na_env_t *env)
{
    if (env->error_count_max > 0) {
        env->error_count++;
    }
}

static void na_connpool_deactivate (na_connpool_t *connpool)
{
    for (int i=0;i<connpool->max;++i) {
        if (connpool->fd_pool[i] > 0) {
            close(connpool->fd_pool[i]);
            connpool->fd_pool[i] = 0;
            connpool->mark[i]    = 0;
        }
    }
}

static bool na_connpool_assign (na_env_t *env, int *cur, int *fd)
{
    na_connpool_t *connpool;

    connpool = NA_CONNPOOL_SELECT(env);

    pthread_mutex_lock(&env->lock_connpool);

    switch (rand() % 2) {
    case 0:
        for (int i=env->connpool_max-1;i>=0;--i) {
            if (connpool->mark[i] == 0) {
                connpool->mark[i] = 1;
                *fd  = connpool->fd_pool[i];
                *cur = i;
                pthread_mutex_unlock(&env->lock_connpool);
                return true;
            }
        }
        break;
    default:
        for (int i=0;i<env->connpool_max;++i) {
            if (connpool->mark[i] == 0) {
                connpool->mark[i] = 1;
                *fd  = connpool->fd_pool[i];
                *cur = i;
                pthread_mutex_unlock(&env->lock_connpool);
                return true;
            }
        }
        break;
    }
    pthread_mutex_unlock(&env->lock_connpool);
    return false;
}

static void na_connpool_init (na_env_t *env)
{
    for (int i=0;i<env->connpool_max;++i) {
        env->connpool_active.fd_pool[i] = na_target_server_tcpsock_init();
        na_target_server_tcpsock_setup(env->connpool_active.fd_pool[i], true);
        if (env->connpool_active.fd_pool[i] <= 0) {
            NA_DIE_WITH_ERROR(NA_ERROR_INVALID_FD);
        }
        
        if (!na_server_connect(env->connpool_active.fd_pool[i], &env->target_server.addr)) {
            if (errno != EINPROGRESS && errno != EALREADY) {
                NA_DIE_WITH_ERROR(NA_ERROR_CONNECTION_FAILED);
            }
        }
    }
}

static void na_connpool_switch (na_env_t *env)
{
    na_connpool_t *connpool;
    na_server_t   *server;
    if (env->is_refused_active) {
        na_connpool_deactivate(&env->connpool_active);
        connpool = &env->connpool_backup;
        server   = &env->backup_server;
    } else {
        na_connpool_deactivate(&env->connpool_backup);
        connpool = &env->connpool_active;
        server   = &env->target_server;
    }

    for (int i=0;i<env->connpool_max;++i) {
        connpool->fd_pool[i] = na_target_server_tcpsock_init();
        na_target_server_tcpsock_setup(connpool->fd_pool[i], true);
        if (connpool->fd_pool[i] <= 0) {
            NA_DIE_WITH_ERROR(NA_ERROR_INVALID_FD);
        }
        
        if (!na_server_connect(connpool->fd_pool[i], &server->addr)) {
            if (errno != EINPROGRESS && errno != EALREADY) {
                NA_DIE_WITH_ERROR(NA_ERROR_CONNECTION_FAILED);
            }
        }
    }
}

static void na_client_close (na_client_t *client, na_env_t *env)
{
    close(client->cfd);
    client->cfd = -1;
    if (client->is_use_connpool) {
        client->connpool->mark[client->cur_pool] = 0;
    } else {
        close(client->tsfd);
    }

    NA_FREE(client->crbuf);
    NA_FREE(client->srbuf);
    NA_FREE(client);

    if (env->current_conn > 0) {
        --env->current_conn;
    }
}

static void na_target_server_callback (EV_P_ struct ev_io *w, int revents)
{
    int cfd, tsfd, size;
    na_client_t *client;
    na_env_t *env;

    tsfd   = w->fd;
    client = (na_client_t *)w->data;
    env    = client->env;
    cfd    = client->cfd;

    if (client->is_refused_active != env->is_refused_active) {
        ev_io_stop(EV_A_ w);
        na_client_close(client, env);
        return; // request fail
    }

    if (env->loop_max > 0 && client->loop_cnt++ > env->loop_max) {
        ev_io_stop(EV_A_ w);
        na_client_close(client, env);
        return; // request fail
    }

    if (revents & EV_READ) {

        if (client->srbufsize >= env->bufsize) {
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            na_error_count_up(env);
            NA_STDERR_MESSAGE(NA_ERROR_OUTOF_BUFFER);
            return; // request fail
        }

        size = read(tsfd, 
                    client->srbuf + client->srbufsize,
                    env->bufsize - client->srbufsize);

        if (size == 0) {
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            return; // request fail
        } else if (size < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                return; // not ready yet
            }
            ev_io_stop(EV_A_ w);
            na_client_close(client, env); // request fail
            return;
        }

        client->srbufsize                += size;
        client->srbuf[client->srbufsize]  = '\0';

        if (client->cmd == NA_MEMPROTO_CMD_GET) {
            client->res_cnt = na_memproto_count_response_get(client->srbuf, client->srbufsize);
            if (client->res_cnt >= client->req_cnt) {
                client->event_state = NA_EVENT_STATE_CLIENT_WRITE;
                na_event_switch(EV_A_ w, &client->c_watcher, cfd, EV_WRITE);
                return;
            }
        } else if (client->srbufsize > 2 &&
                   client->srbuf[client->srbufsize - 2] == '\r' &&
                   client->srbuf[client->srbufsize - 1] == '\n')
        {
            client->event_state = NA_EVENT_STATE_CLIENT_WRITE;
            na_event_switch(EV_A_ w, &client->c_watcher, cfd, EV_WRITE);
        }

    } else if (revents & EV_WRITE) {

        size = write(tsfd, 
                     client->crbuf + client->swbufsize, 
                     client->crbufsize - client->swbufsize);

        if (size < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
                return; // not ready yet
            } else if (client->is_use_connpool) {
                NA_STDERR_MESSAGE(NA_ERROR_UNKNOWN);
                int i = client->cur_pool;
                na_server_t *server;
                if (client->connpool->fd_pool[i] > 0) {
                    close(client->connpool->fd_pool[i]);
                }
                client->connpool->fd_pool[i] = na_target_server_tcpsock_init();
                server = env->is_refused_active ? &env->backup_server : &env->target_server;
                na_target_server_tcpsock_setup(client->connpool->fd_pool[i], true);
                if (client->connpool->fd_pool[i] <= 0) {
                    NA_DIE_WITH_ERROR(NA_ERROR_INVALID_FD);
                }
                
                if (!na_server_connect(client->connpool->fd_pool[i], &server->addr)) {
                    if (errno != EINPROGRESS && errno != EALREADY) {
                        na_error_count_up(env);
                        NA_DIE_WITH_ERROR(NA_ERROR_CONNECTION_FAILED);
                    }
                }
            }
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            return; // request fail
        }

        client->swbufsize += size;

        if (client->swbufsize < client->crbufsize) {
            na_event_switch(EV_A_ w, &client->ts_watcher, tsfd, EV_WRITE);
        } else {
            client->event_state = NA_EVENT_STATE_TARGET_READ;
            na_event_switch(EV_A_ w, &client->ts_watcher, tsfd, EV_READ);
        }
        return;
    }

}

static void na_client_callback(EV_P_ struct ev_io *w, int revents)
{
    int cfd, tsfd, size;
    na_client_t *client;
    na_env_t *env;

    cfd    = w->fd;
    client = (na_client_t *)w->data;
    env    = client->env;
    tsfd   = client->tsfd;

    if (client->is_refused_active != env->is_refused_active) {
        ev_io_stop(EV_A_ w);
        na_client_close(client, env);
        return; // request fail
    }

    if (env->loop_max > 0 && client->loop_cnt++ > env->loop_max) {
        ev_io_stop(EV_A_ w);
        na_client_close(client, env);
        return; // request fail
    }

    if (revents & EV_READ) {

        size = read(cfd, 
                    client->crbuf + client->crbufsize,
                    env->bufsize - client->crbufsize);

        if (size == 0) {
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            return; // request success
        } else if (size < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                return; // not ready yet
            }
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            return; // request fail
        }

        client->crbufsize                += size;
        client->crbuf[client->crbufsize]  = '\0';

        client->cmd = na_memproto_detect_command(client->crbuf);

        if (client->cmd == NA_MEMPROTO_CMD_QUIT) {
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            return; // request success
        } else if (client->cmd == NA_MEMPROTO_CMD_GET) {
            client->req_cnt = na_memproto_count_request_get(client->crbuf, client->crbufsize);
        }

        if (client->crbufsize < 2) {
            return; // not ready yet
        } else if (client->crbuf[client->crbufsize - 2] == '\r' &&
                   client->crbuf[client->crbufsize - 1] == '\n')
        {
            if (client->cmd == NA_MEMPROTO_CMD_UNKNOWN) {
                ev_io_stop(EV_A_ w);
                na_client_close(client, env);
                return; // request fail
            }
            client->event_state = NA_EVENT_STATE_TARGET_WRITE;
            ev_io_stop(EV_A_ w);
            ev_io_init(&client->ts_watcher, na_target_server_callback, tsfd, EV_WRITE);
            ev_io_start(EV_A_ &client->ts_watcher);
            return;
        }

    } else if (revents & EV_WRITE) {

        size = write(cfd,
                     client->srbuf + client->cwbufsize,
                     client->srbufsize - client->cwbufsize);

        if (size < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
                return; // not ready yet
            }
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            return; // request fail
        }

        client->cwbufsize += size;
        if (client->cwbufsize < client->srbufsize) {
            na_event_switch(EV_A_ w, &client->c_watcher, cfd, EV_WRITE);
            return;
        } else {
            client->crbufsize   = 0;
            client->cwbufsize   = 0;
            client->srbufsize   = 0;
            client->swbufsize   = 0;
            client->event_state = NA_EVENT_STATE_CLIENT_READ;
            client->req_cnt     = 0;
            client->res_cnt     = 0;
            na_event_switch(EV_A_ w, &client->c_watcher, cfd, EV_READ);
            return;
        }
        
    }
}

void na_front_server_callback (EV_P_ struct ev_io *w, int revents)
{
    int fsfd, cfd, tsfd, cur_pool;
    int th_ret;<commit_msg>insert error count function when errro occured<commit_after>/**
   In short, neoagent is distributed under so called "BSD license",
   
   Copyright (c) 2011-2012 <NAME> <<EMAIL>>
   All rights reserved.
   
   Redistribution and use in source and binary forms, with or without modification, 
   are permitted provided that the following conditions are met:
   
   * Redistributions of source code must retain the above copyright notice, 
   this list of conditions and the following disclaimer.
   
   * Redistributions in binary form must reproduce the above copyright notice, 
   this list of conditions and the following disclaimer in the documentation 
   and/or other materials provided with the distribution.
   
   * Neither the name of the authors nor the names of its contributors 
   may be used to endorse or promote products derived from this software 
   without specific prior written permission.
   
   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 
   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 
   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 
   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED 
   TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
   PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
   LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
   NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
   SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdbool.h>
#include <stdint.h>
#include <unistd.h>
#include <errno.h>
#include <sys/ioctl.h>
#include <pthread.h>
#include <assert.h>
#include <time.h>

#include <ev.h>

#include "event.h"
#include "env.h"
#include "socket.h"
#include "error.h"
#include "memproto.h"
#include "stat.h"
#include "util.h"

#define NA_CONNPOOL_SELECT(env) (env->is_refused_active ? &env->connpool_backup : &env->connpool_active)

// constants
static const int NA_STAT_BUF_MAX = 8192;

// external globals
volatile sig_atomic_t SigExit;

// private functions
inline static void na_event_switch (EV_P_ struct ev_io *old, ev_io *new, int fd, int revent);
inline static void na_error_count_up (na_env_t *env);

static void na_connpool_deactivate (na_connpool_t *connpool);
static bool na_connpool_assign (na_env_t *env, int *cur, int *fd);
static void na_connpool_init (na_env_t *env);
static void na_connpool_switch (na_env_t *env);
static void na_client_close (na_client_t *client, na_env_t *env);
static void na_health_check_callback (EV_P_ ev_timer *w, int revents);
static void na_stat_callback (EV_P_ struct ev_io *w, int revents);
static void na_target_server_callback (EV_P_ struct ev_io *w, int revents);
static void na_client_callback (EV_P_ struct ev_io *w, int revents);
static void na_front_server_callback (EV_P_ struct ev_io *w, int revents);
static void *na_support_loop (void *args);

inline static void na_event_switch (EV_P_ struct ev_io *old, struct ev_io *new, int fd, int revent)
{
    ev_io_stop(EV_A_ old);
    ev_io_set(new, fd, revent);
    ev_io_start(EV_A_ new);
}

inline static void na_error_count_up (na_env_t *env)
{
    if (env->error_count_max > 0) {
        env->error_count++;
    }
}

static void na_connpool_deactivate (na_connpool_t *connpool)
{
    for (int i=0;i<connpool->max;++i) {
        if (connpool->fd_pool[i] > 0) {
            close(connpool->fd_pool[i]);
            connpool->fd_pool[i] = 0;
            connpool->mark[i]    = 0;
        }
    }
}

static bool na_connpool_assign (na_env_t *env, int *cur, int *fd)
{
    na_connpool_t *connpool;

    connpool = NA_CONNPOOL_SELECT(env);

    pthread_mutex_lock(&env->lock_connpool);

    switch (rand() % 2) {
    case 0:
        for (int i=env->connpool_max-1;i>=0;--i) {
            if (connpool->mark[i] == 0) {
                connpool->mark[i] = 1;
                *fd  = connpool->fd_pool[i];
                *cur = i;
                pthread_mutex_unlock(&env->lock_connpool);
                return true;
            }
        }
        break;
    default:
        for (int i=0;i<env->connpool_max;++i) {
            if (connpool->mark[i] == 0) {
                connpool->mark[i] = 1;
                *fd  = connpool->fd_pool[i];
                *cur = i;
                pthread_mutex_unlock(&env->lock_connpool);
                return true;
            }
        }
        break;
    }
    pthread_mutex_unlock(&env->lock_connpool);
    return false;
}

static void na_connpool_init (na_env_t *env)
{
    for (int i=0;i<env->connpool_max;++i) {
        env->connpool_active.fd_pool[i] = na_target_server_tcpsock_init();
        na_target_server_tcpsock_setup(env->connpool_active.fd_pool[i], true);
        if (env->connpool_active.fd_pool[i] <= 0) {
            NA_DIE_WITH_ERROR(NA_ERROR_INVALID_FD);
        }
        
        if (!na_server_connect(env->connpool_active.fd_pool[i], &env->target_server.addr)) {
            if (errno != EINPROGRESS && errno != EALREADY) {
                NA_DIE_WITH_ERROR(NA_ERROR_CONNECTION_FAILED);
            }
        }
    }
}

static void na_connpool_switch (na_env_t *env)
{
    na_connpool_t *connpool;
    na_server_t   *server;
    if (env->is_refused_active) {
        na_connpool_deactivate(&env->connpool_active);
        connpool = &env->connpool_backup;
        server   = &env->backup_server;
    } else {
        na_connpool_deactivate(&env->connpool_backup);
        connpool = &env->connpool_active;
        server   = &env->target_server;
    }

    for (int i=0;i<env->connpool_max;++i) {
        connpool->fd_pool[i] = na_target_server_tcpsock_init();
        na_target_server_tcpsock_setup(connpool->fd_pool[i], true);
        if (connpool->fd_pool[i] <= 0) {
            NA_DIE_WITH_ERROR(NA_ERROR_INVALID_FD);
        }
        
        if (!na_server_connect(connpool->fd_pool[i], &server->addr)) {
            if (errno != EINPROGRESS && errno != EALREADY) {
                NA_DIE_WITH_ERROR(NA_ERROR_CONNECTION_FAILED);
            }
        }
    }
}

static void na_client_close (na_client_t *client, na_env_t *env)
{
    close(client->cfd);
    client->cfd = -1;
    if (client->is_use_connpool) {
        client->connpool->mark[client->cur_pool] = 0;
    } else {
        close(client->tsfd);
    }

    NA_FREE(client->crbuf);
    NA_FREE(client->srbuf);
    NA_FREE(client);

    if (env->current_conn > 0) {
        --env->current_conn;
    }
}

static void na_target_server_callback (EV_P_ struct ev_io *w, int revents)
{
    int cfd, tsfd, size;
    na_client_t *client;
    na_env_t *env;

    tsfd   = w->fd;
    client = (na_client_t *)w->data;
    env    = client->env;
    cfd    = client->cfd;

    if (client->is_refused_active != env->is_refused_active) {
        ev_io_stop(EV_A_ w);
        na_client_close(client, env);
        na_error_count_up(env);
        return; // request fail
    }

    if (env->loop_max > 0 && client->loop_cnt++ > env->loop_max) {
        ev_io_stop(EV_A_ w);
        na_client_close(client, env);
        na_error_count_up(env);
        return; // request fail
    }

    if (revents & EV_READ) {

        if (client->srbufsize >= env->bufsize) {
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            na_error_count_up(env);
            NA_STDERR_MESSAGE(NA_ERROR_OUTOF_BUFFER);
            return; // request fail
        }

        size = read(tsfd, 
                    client->srbuf + client->srbufsize,
                    env->bufsize - client->srbufsize);

        if (size == 0) {
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            na_error_count_up(env);
            return; // request fail
        } else if (size < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                return; // not ready yet
            }
            ev_io_stop(EV_A_ w);
            na_client_close(client, env); // request fail
            na_error_count_up(env);
            return;
        }

        client->srbufsize                += size;
        client->srbuf[client->srbufsize]  = '\0';

        if (client->cmd == NA_MEMPROTO_CMD_GET) {
            client->res_cnt = na_memproto_count_response_get(client->srbuf, client->srbufsize);
            if (client->res_cnt >= client->req_cnt) {
                client->event_state = NA_EVENT_STATE_CLIENT_WRITE;
                na_event_switch(EV_A_ w, &client->c_watcher, cfd, EV_WRITE);
                return;
            }
        } else if (client->srbufsize > 2 &&
                   client->srbuf[client->srbufsize - 2] == '\r' &&
                   client->srbuf[client->srbufsize - 1] == '\n')
        {
            client->event_state = NA_EVENT_STATE_CLIENT_WRITE;
            na_event_switch(EV_A_ w, &client->c_watcher, cfd, EV_WRITE);
        }

    } else if (revents & EV_WRITE) {

        size = write(tsfd, 
                     client->crbuf + client->swbufsize, 
                     client->crbufsize - client->swbufsize);

        if (size < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
                return; // not ready yet
            } else if (client->is_use_connpool) {
                NA_STDERR_MESSAGE(NA_ERROR_UNKNOWN);
                int i = client->cur_pool;
                na_server_t *server;
                if (client->connpool->fd_pool[i] > 0) {
                    close(client->connpool->fd_pool[i]);
                }
                client->connpool->fd_pool[i] = na_target_server_tcpsock_init();
                server = env->is_refused_active ? &env->backup_server : &env->target_server;
                na_target_server_tcpsock_setup(client->connpool->fd_pool[i], true);
                if (client->connpool->fd_pool[i] <= 0) {
                    NA_DIE_WITH_ERROR(NA_ERROR_INVALID_FD);
                }
                
                if (!na_server_connect(client->connpool->fd_pool[i], &server->addr)) {
                    if (errno != EINPROGRESS && errno != EALREADY) {
                        na_error_count_up(env);
                        NA_DIE_WITH_ERROR(NA_ERROR_CONNECTION_FAILED);
                    }
                }
            }
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            na_error_count_up(env);
            return; // request fail
        }

        client->swbufsize += size;

        if (client->swbufsize < client->crbufsize) {
            na_event_switch(EV_A_ w, &client->ts_watcher, tsfd, EV_WRITE);
        } else {
            client->event_state = NA_EVENT_STATE_TARGET_READ;
            na_event_switch(EV_A_ w, &client->ts_watcher, tsfd, EV_READ);
        }
        return;
    }

}

static void na_client_callback(EV_P_ struct ev_io *w, int revents)
{
    int cfd, tsfd, size;
    na_client_t *client;
    na_env_t *env;

    cfd    = w->fd;
    client = (na_client_t *)w->data;
    env    = client->env;
    tsfd   = client->tsfd;

    if (client->is_refused_active != env->is_refused_active) {
        ev_io_stop(EV_A_ w);
        na_client_close(client, env);
        na_error_count_up(env);
        return; // request fail
    }

    if (env->loop_max > 0 && client->loop_cnt++ > env->loop_max) {
        ev_io_stop(EV_A_ w);
        na_client_close(client, env);
        na_error_count_up(env);
        return; // request fail
    }

    if (revents & EV_READ) {

        size = read(cfd, 
                    client->crbuf + client->crbufsize,
                    env->bufsize - client->crbufsize);

        if (size == 0) {
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            return; // request success
        } else if (size < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                return; // not ready yet
            }
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            na_error_count_up(env);
            return; // request fail
        }

        client->crbufsize                += size;
        client->crbuf[client->crbufsize]  = '\0';

        client->cmd = na_memproto_detect_command(client->crbuf);

        if (client->cmd == NA_MEMPROTO_CMD_QUIT) {
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            return; // request success
        } else if (client->cmd == NA_MEMPROTO_CMD_GET) {
            client->req_cnt = na_memproto_count_request_get(client->crbuf, client->crbufsize);
        }

        if (client->crbufsize < 2) {
            return; // not ready yet
        } else if (client->crbuf[client->crbufsize - 2] == '\r' &&
                   client->crbuf[client->crbufsize - 1] == '\n')
        {
            if (client->cmd == NA_MEMPROTO_CMD_UNKNOWN) {
                ev_io_stop(EV_A_ w);
                na_client_close(client, env);
                na_error_count_up(env);
                return; // request fail
            }
            client->event_state = NA_EVENT_STATE_TARGET_WRITE;
            ev_io_stop(EV_A_ w);
            ev_io_init(&client->ts_watcher, na_target_server_callback, tsfd, EV_WRITE);
            ev_io_start(EV_A_ &client->ts_watcher);
            return;
        }

    } else if (revents & EV_WRITE) {

        size = write(cfd,
                     client->srbuf + client->cwbufsize,
                     client->srbufsize - client->cwbufsize);

        if (size < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
                return; // not ready yet
            }
            ev_io_stop(EV_A_ w);
            na_client_close(client, env);
            na_error_count_up(env);
            return; // request fail
        }

        client->cwbufsize += size;
        if (client->cwbufsize < client->srbufsize) {
            na_event_switch(EV_A_ w, &client->c_watcher, cfd, EV_WRITE);
            return;
        } else {
            client->crbufsize   = 0;
            client->cwbufsize   = 0;
            client->srbufsize   = 0;
            client->swbufsize   = 0;
            client->event_state = NA_EVENT_STATE_CLIENT_READ;
            client->req_cnt     = 0;
            client->res_cnt     = 0;
            na_event_switch(EV_A_ w, &client->c_watcher, cfd, EV_READ);
            return;
        }
        
    }
}

void na_front_server_callback (EV_P_ struct ev_io *w, int revents)
{
    int fsfd, cfd, tsfd, cur_pool;
    int th_ret;
{'fraction_non_alphanumeric': 0.11150644236415766, 'fraction_numerical': 0.003937959860444229, 'mean_word_length': 2.1072233551572395, 'pattern_counts': {'":': 0, '<': 61, '<?xml version=': 0, '>': 379, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2527802', 'n_tokens_mistral': 9806, 'n_tokens_neox': 9379, 'n_words': 2065}