/*********************************************************************************
 *      Copyright:  (C) 2012 Guo Wenxue<guowenxue@gmail.com>  
 *                  All rights reserved.
 *
 *       Filename:  cp_fds.c
 *    Description:  This file is the linux epoll basic library
 *                 
 *        Version:  1.0.0(10/25/2012~)
 *         Author:  Guo Wenxue <guowenxue@gmail.com>
 *      ChangeLog:  1, Release initial version on "10/25/2012 04:55:30 PM"
 *                 
 ********************************************************************************/

#include <sys/resource.h>
#include "cp_fds.h"
#include "cp_sock.h"
#include "cp_time.h"
#include "cp_common.h"

/* set maximum file descriptor number that can be opened in this process */
int cp_fds_setrlimit(int maxfd)
{
    struct  rlimit    res;

    if(getrlimit(RLIMIT_NOFILE, &res) <0)
    {
        log_fatal("getrlimit failed: %s\n", strerror(errno));
        return -1;
    }

    if(res.rlim_cur < maxfd)
    {
        res.rlim_cur = maxfd;
        if(setrlimit(RLIMIT_NOFILE, &res) <0)
        {
            log_fatal("setrlimit failed: %s\n", strerror(errno));
            return -2;
        }
    }

    return 0;
}

/* Initialze $fds context, if $fds is NULL, then malloc it */
CP_FDS *cp_fds_init(CP_FDS *fds, int maxevents, int timeout)
{
    int          rv = 0;

    if(NULL != fds)
    { 
        memset(fds, 0, sizeof(*fds));
        fds->epfd = -1;
    }
    else
    {
        if( !(fds=(CP_FDS *)t_malloc(sizeof(*fds))) )
        {
            log_err("fds context malloc failed: %s\n", strerror(errno));
            return NULL;
        }
        else
        {
            log_trace("malloc fds with address [%p]\n", fds);
            memset(fds, 0, sizeof(*fds));
            fds->epfd = -1;
            fds->flag |= FLAG_FDS_MALLOC;
        }
    }

    /* set maximum file descriptor number that can be opened in this process */
    maxevents = maxevents<CP_DEF_MAX_EVENTS ? maxevents : CP_DEF_MAX_EVENTS;
    if( cp_fds_setrlimit(maxevents) )
    {
        rv = -2;
        goto cleanup;
    }

    /* Create epoll file description */
    fds->epfd = epoll_create(maxevents);
    if(fds->epfd < 0)
    {
        rv = -3;
        log_fatal("epoll_create failed: %s\n", strerror(errno));
        goto cleanup;
    }
    log_trace("Open epoll file description [%d]\n", fds->epfd);

    /* Initialise the server/client/task list */
    INIT_LIST_HEAD(&fds->server_list);
    INIT_LIST_HEAD(&fds->client_list);
    fds->task_array = cp_array_init(NULL, maxevents);
    fds->event_queue = cp_queue_init(NULL, maxevents);

    fds->timeout = timeout<0 ? -1 : timeout;
    fds->max_event = maxevents;
    fds->flag |= FLAG_FDS_INIT;

cleanup:
    if(rv)
    {
        log_err("Initialise fds contex failed\n");
        cp_fds_term(fds);
        return NULL;
    }

    log_nrml("Initialise fds contex [%p] ok\n", fds);
    return fds;
}

/* Terminate $fds context, if $fds malloced, then free it */
void cp_fds_term(CP_FDS *fds)
{
    log_dbg("terminate epoll fds contex now\n");

    if(!fds)
    {
        log_err("Invalude input arguments\n");
        return;
    }

    if(fds->epfd >= 0)
    {
        log_dbg("Close epoll file description [%d]\n", fds->epfd); 
        close(fds->epfd); 
    }

    if(fds->task_array)
    {
        cp_array_term(fds->task_array);
    }

    if(fds->event_queue)
    {
        cp_queue_destroy(fds->event_queue);
    }

    if(fds->flag& FLAG_FDS_MALLOC)
    {
        t_free(fds);
    }

    log_nrml("Terminate epoll fds contex ok.\n");

    return ;
}

/* epoll_ctl() to add this socket to epoll   */
int cp_add_epoll_event(CP_SOCK *sock)
{
    int                     rv = 0;
    CP_FDS                  *fds;

    if( !sock || !(fds=sock->fds) )
    {
        log_err("Invalude input arguments\n");
        return -1;
    }

    if( sock->flag&FLAG_SOCK_EPOLLED )
        return 0; /* already put in epoll */

    if(fds->event_cnt > fds->max_event)
    {
        log_fatal("No more epoll event space [%d/%d] for socket[%d]\n", 
                fds->event_cnt, fds->max_event, sock->fd);
        return -2;
    }

    memset(&sock->event, 0, sizeof(sock->event));
    sock->event.events = EPOLLIN;
    sock->event.data.ptr = sock;

    if ( epoll_ctl (fds->epfd, EPOLL_CTL_ADD, sock->fd, &sock->event) < 0) 
    {
        if(EEXIST == errno)
        {
            log_warn("socket[%d] already registe in epoll\n", sock->fd);
            goto cleanup;
        }
        else
        {
            log_err("socket[%d] registe in epoll failed: %s\n", sock->fd, strerror(errno));
            rv = -3;
            goto cleanup;
        }
    }
    
    fds->event_cnt++;
    sock->flag |= FLAG_SOCK_EPOLLED;

cleanup:
    if(rv)
        log_err("add socket [%d] to epoll event failed\n", sock->fd);
    else
        log_dbg("add socket [%d] to epoll event ok\n", sock->fd);

    return rv;
}


/* epoll_ctl() to mod this socket in epoll */
int cp_mod_epoll_event(CP_SOCK *sock, int event)
{
    CP_FDS                  *fds;

    if( !sock || !(fds=sock->fds) )
    {
        log_err("Invalude input arguments\n");
        return -1;
    }

    if( !(sock->flag&FLAG_SOCK_EPOLLED) )
    {
        return -2;  /* not in epoll  */
    }

    sock->event.events = event;
    if ( epoll_ctl (fds->epfd, EPOLL_CTL_MOD, sock->fd, &sock->event) < 0) 
    {
        log_err("modidfy socket [%d] epoll event failed: %s\n", sock->fd, strerror(errno));
        return -3;
    }
    
    log_dbg("modidfy socket [%d] epoll event ok\n", sock->fd);

    return 0;
}


void cp_del_epoll_event(CP_SOCK *sock)
{
    CP_FDS                  *fds;

    if( !sock || !(fds=sock->fds) )
    {
        log_err("Invalude input arguments\n");
        return;
    }

    if( !(sock->flag&FLAG_SOCK_EPOLLED) )
        return; /* not in epoll  */

    sock->flag &= ~FLAG_SOCK_EPOLLED;
    epoll_ctl(fds->epfd, EPOLL_CTL_DEL, sock->fd, NULL);
    fds->event_cnt--;

    log_dbg("remove socket [%d] from epoll event ok\n", sock->fd);
    return ;
}

/* add the listen/connect socket into server_list/client_list */
int cp_fds_add_sock_registry(CP_SOCK *sock)
{
    CP_FDS                  *fds;

    if( !sock || !(fds=sock->fds) )
    {
        log_err("Invalude input arguments\n");
        return -1;
    }

    if( sock->flag&FLAG_SOCK_REGISTRY )
        return 0;  /* already in registry list  */

    if(CP_SOCK_MODE_CONNECT == sock->mode)
    {
        log_dbg("regist socket [%p] on CONNECT mode into client list ok\n", sock);
        list_add_tail(&sock->rlist, &fds->client_list);
    }
    else if(CP_SOCK_MODE_LISTEN == sock->mode)
    {
        log_dbg("regist socket [%p] on LISTEN mode into server list ok\n", sock);
        list_add_tail(&sock->rlist, &fds->server_list);
    }
    else if(CP_SOCK_MODE_ACCEPT==sock->mode && sock->serv_sock)
    {
        log_dbg("regist socket [%p] on ACCEPT mode into server list ok\n", sock);
        list_add_tail(&sock->rlist, &sock->serv_sock->accept_list);
        sock->serv_sock->accept_cnt++;
    }
    else
    {
        log_err("regist socket [%p] on mode %d into client/server list failed: Unsupport mode.\n", sock, sock->mode);
        return -2;
    }

    sock->flag |= FLAG_SOCK_REGISTRY;
    return 0;
}

/*  remove the listen/connect socket from server_list/client_list */
void cp_fds_del_sock_registry(CP_SOCK *sock)
{
    if(!sock || !(sock->flag&FLAG_SOCK_REGISTRY) )
        return;  /* not in registry list  */

    if(CP_SOCK_MODE_ACCEPT==sock->mode && sock->serv_sock)
        sock->serv_sock->accept_cnt--;

    log_dbg("remove socket [%d] from socket registry list ok\n", sock->fd);
    list_del(&sock->rlist);
    sock->flag &= ~FLAG_SOCK_REGISTRY;

    return ;
}


/* Add a socket in task list */
int cp_fds_add_sock_task(CP_SOCK *sock)
{
    CP_FDS                  *fds;
    int                     rv;

    if( !sock || !(fds=sock->fds) )
    {
        log_err("Invalude input arguments\n");
        return -1;
    }

    if( sock->flag&FLAG_SOCK_TASKED )
        return 0; /* already in a list */

    log_info("add socket [%d] into task list\n", sock->fd);
    if( (rv=cp_array_add(fds->task_array, sock)) >=0 )
    {
        sock->index = rv;
        sock->flag |= FLAG_SOCK_TASKED;
        return 0;
    }

    return -2;
}

void cp_fds_del_sock_task(CP_SOCK *sock)
{
    if(!sock || !(sock->flag&FLAG_SOCK_TASKED))
        return ;
 
    log_info("remove socket [%d:%p] from task list[%p] by index [%d]\n", sock->fd, sock, sock->fds->task_array, sock->index);

    if(sock->index >= 0)
        cp_array_rm_byindex(sock->fds->task_array, sock->index);
    else
        cp_array_rm_bydata(sock->fds->task_array, sock);

    sock->flag &= ~FLAG_SOCK_TASKED;
    sock->index = -1;
    memset(&sock->event, 0, sizeof(struct epoll_event));

    return ;
}

void *cp_fds_sock_enqueue(CP_SOCK *sock)
{
    if(!sock || sock->flag&FLAG_SOCK_INQUEUE || sock->flag&FLAG_SOCK_TASKED || !sock->fds)
        return NULL;

    sock->flag |= FLAG_SOCK_INQUEUE;
    return cp_enqueue(sock->fds->event_queue, sock);
}

void *cp_fds_sock_dequeue(CP_FDS *fds)
{
    CP_SOCK    *sock = NULL;

    if( !fds )
        return NULL;

    sock = cp_dequeue(fds->event_queue);
    sock->flag &= ~FLAG_SOCK_INQUEUE;

    return sock;
}

void *cp_fds_sock_rmqueue(CP_SOCK *sock)
{
    if(!sock || !(sock->flag&FLAG_SOCK_INQUEUE) || !sock->fds)
        return NULL;

    sock->flag &= ~FLAG_SOCK_INQUEUE;
    return cp_rmqueue(sock->fds->event_queue, sock);
}

void cp_fds_list_sock_task(CP_FDS *fds)
{
    int               i;
    CP_SOCK           *sock;

    if(!fds)
        return ;

    log_dbg("list all the socket in task list:\n");

    /* list all the socket from task list */
    cp_list_array_for_each(fds->task_array, i, sock)
    { 
        if(sock)
            log_dbg("socket[%d:%p] in task list\n", sock->fd, sock); 
    }

    return ;
}


void cp_fds_destroy_sock(CP_SOCK *sock)
{
    cp_fds_del_sock_task(sock);
    cp_fds_sock_rmqueue(sock);
    cp_del_epoll_event(sock);
    cp_fds_del_sock_registry(sock);
    cp_sock_term(sock);

    return ;
}


int cp_fds_detect_event(CP_FDS *fds)
{
    int                   i=0, nfds=0;
    CP_SOCK               *sock;
    struct epoll_event    evts[CP_MAX_EVENT_ONCE];

    if( !fds )
        return -1;

    nfds = epoll_wait(fds->epfd, evts, CP_MAX_EVENT_ONCE, fds->timeout);
    ///if(nfds<0 && errno!=EINTR)
    if(nfds<0)
    {
        log_fatal("epoll_wait failure: %s\n", strerror(errno));
        return -2;
    }

    if(nfds == 0)
        return 0;

    log_dbg("epoll_wait get [%d] events\n", nfds);

    for (i=0; i<nfds; i++) 
    {
        sock=(CP_SOCK *)evts[i].data.ptr;
        sock->event = evts[i];
        sock->actv_time = time_now();

        log_dbg("socket [%d] get event [0x%0x] and added to event queue ok\n", sock->fd, sock->event.events);
        if( SOCK_STAT_ALREADY==sock->status )
            cp_fds_sock_enqueue(sock);
    }

    return 0;
}

void cp_fds_proc_event(CP_FDS *fds)
{
    int                    i;
    CP_SOCK                *sock;
    int                    rv;

    if( !fds )
        return;

    //cp_fds_list_sock_task(fds);

    //log_trace("Start process the events....\n");
    
    /* Process all the socket in task list first, its urgent */
    cp_list_array_for_each(fds->task_array, i, sock)
    {
        if(sock)
        {
            if(sock->cbfunc)
            {
                log_dbg("Process socket [%d] in task list on event [%d]\n", sock->fd, sock->event.events);
                rv = sock->cbfunc(sock);
            }
            else
            { 
                log_err("Process Socket [%d:%p] not implement event callback and remove it\n", sock->fd, sock);
                cp_fds_del_sock_task(sock);
            }
        }
    }

    /* Process all the socket in event queue now */
    while(!cp_queue_is_empty(fds->event_queue))
    {
        if( NULL!= (sock=cp_fds_sock_dequeue(fds)) )
        { 
            //log_dbg("after handle one socket count: %d\n", cp_queue_count(fds->event_queue));
            if(sock->cbfunc)
            { 
                log_dbg("Process socket [%d %p] in event queue on event [%d]\n", sock->fd, sock, sock->event.events); 
                rv = sock->cbfunc(sock);
            }
            else
            { 
                log_err("Process Socket [%d:%p] not implement event callback and remove it\n", sock->fd, sock);
            } 
        }
    }

    //log_trace("End process the events....\n");
    return;
}


void cp_sock_term_server(CP_SOCK *serv_sock)
{
    CP_SOCK           *sock, *tsock;

    log_warn("terminate server socket [%d] and [%d] accept client now\n", serv_sock->fd, serv_sock->accept_cnt);
    list_for_each_entry_safe(sock, tsock, &serv_sock->accept_list, rlist) 
    {
        /*remove all the accept socket from task and registry list, then destroy it*/
        //log_warn("destroy accept socket [%d:%p]\n", sock->fd, sock);
        cp_fds_destroy_sock(sock);
    }

    /*remove this server socket from task and registry list, then destroy it*/
    log_warn("terminate server socket [%d] and its accept client ok\n", serv_sock->fd);
    cp_fds_destroy_sock(serv_sock);

    return ;
}


void cp_sock_term_all_server(CP_FDS *fds)
{ 
    CP_SOCK           *sock, *tsock;

    log_dbg("destroy all the listen socket now\n");

    /* Terminate all the listen socket */
    list_for_each_entry_safe(sock, tsock, &fds->server_list, rlist) 
    {
        cp_sock_term_server(sock);
    }

    log_warn("destroy all the listen socket ok\n");

    return ;
}


void cp_sock_term_all_client(CP_FDS *fds)
{
    CP_SOCK           *sock = NULL, *tsock;

    log_dbg("destroy all the connect socket now\n");
    /* Terminate all the listen socket */
    list_for_each_entry_safe(sock, tsock, &fds->client_list, rlist) 
    {
        /*remove all the connect socket from task and registry list, then destroy it*/
        cp_fds_destroy_sock(sock);
    }
    log_warn("destroy all the connect socket ok\n");
}

/* Checkout the socket timeout happened or not, if happened then put it in task list */
void cp_sock_detect_timeout(CP_FDS *fds)
{ 
    CP_SOCK           *sock, *tsock, *serv_sock;

    /* check all the connect socket timeout */
    list_for_each_entry_safe(sock, tsock, &fds->client_list, rlist) 
    {
        /*If the socket timeout value and timeout, then disconnect it */
        if( sock->idle_timeout>0 && time_elapsed(sock->actv_time)>sock->idle_timeout )
        {
            log_warn("socket[%d] idle timeout happened and add to task list\n", sock->fd);
            log_dbg("last: %lu elapsed time: %lu idel timeout: %lu\n", sock->actv_time, time_elapsed(sock->actv_time), sock->idle_timeout);
            sock->event.events = CP_SOCK_EVENT_IDLE_TIMEOUT;
            cp_fds_add_sock_task(sock);
        }
        else if( sock->msg_timeout>0 && sock->msg_time>0 && time_elapsed(sock->msg_time)>sock->msg_timeout )
        {
            log_warn("socket[%d] message timeout happened and add to task list\n", sock->fd);
            sock->event.events = CP_SOCK_EVENT_MSG_TIMEOUT;
            cp_fds_add_sock_task(sock);
        }
    } 

    /* check all the accept socket timeout */
    list_for_each_entry(serv_sock, &fds->server_list, rlist) 
    {
        list_for_each_entry_safe(sock, tsock, &serv_sock->accept_list, rlist) 
        {
            /*If the socket timeout value and timeout, then disconnect it */
            if( sock->idle_timeout>0 && time_elapsed(sock->actv_time)>sock->idle_timeout )
            {
                sock->event.events = CP_SOCK_EVENT_IDLE_TIMEOUT;
                log_warn("socket[%d] idle timeout happened and add to task list\n", sock->fd);
                log_dbg("last: %lu elapsed time: %lu idel timeout: %lu\n", sock->actv_time, time_elapsed(sock->actv_time), sock->idle_timeout);
                cp_fds_add_sock_task(sock);
            }
            else if( sock->msg_timeout>0 && sock->msg_time>0 && time_elapsed(sock->msg_time)>sock->msg_timeout )
            {
                sock->event.events = CP_SOCK_EVENT_MSG_TIMEOUT;
                log_warn("socket[%d] message timeout happened and add to task list\n", sock->fd);
                cp_fds_add_sock_task(sock); 
            }
        } /* list_for_each_entry_safe server accept socket list */
    } /* list_for_each_entry all the server socket list */
}


void cp_sock_term_all_task(CP_FDS *fds)
{
    int               i;
    CP_SOCK           *sock;

    log_dbg("remove all the socket in task list now\n");

    /* remove all the socket from task array */
    cp_list_array_for_each(fds->task_array, i, sock)
    {
        //log_warn("remove socket[%d] in task list ok\n", sock->fd);
        if(sock)
            cp_fds_del_sock_task(sock);
    }

    while( !cp_queue_is_empty(fds->event_queue) )
    {
        cp_dequeue(fds->event_queue);
    }

    log_warn("remove all the socket in task list ok\n");
    
    return ;
}