#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <stdint.h>
#include <pthread.h>
#include <time.h>
#include <linux/netlink.h>
#include <semaphore.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/sysinfo.h>


#include "libnetlinku.h"
//#include "policy_client.h"
//#include "policy_common.h"
#include "log.h"
#include "collection.h"
#include "trace_msg.h"
#include "commuapinl.h"

#define HASH_SESS_TAB_BITS     8
#define HASH_SESS_TAB_SIZE     (1 << HASH_SESS_TAB_BITS)

#define EXEC_SYNC_WAIT_TIMEOUT      5

#define MAX_QUEUE_COUNT             256
#define MAX_QUEUE_TIMEOUT           (EXEC_SYNC_WAIT_TIMEOUT + 2)

#define TRACE_POLICY_BUF_SZ        (sizeof(struct nlmsghdr) + TRACE_REQ_SZ)

#define SESS_HASH_INDEX(seq)    (seq >> HASH_SESS_TAB_BITS)

typedef struct _cb_arg {
    struct hlist_node   node;
    uint32_t            seq;
    void                *arg;  
    long                t;      // 加入链表的时间
    void                (*cb)(trace_ret_t ret, void *arg);
} cb_arg_t;

typedef struct _sync_arg {
    trace_ret_t ret;
    sem_t    sem;
} sync_arg_t;


typedef struct _sess {
    uint16_t            count[HASH_SESS_TAB_SIZE];
    struct hlist_head   hess[HASH_SESS_TAB_SIZE];
    struct hlist_node   *last[HASH_SESS_TAB_SIZE];
    pthread_mutex_t     hsess_mutex[HASH_SESS_TAB_SIZE];
} sess_t;

static int g_pid;
static pthread_t g_client_thread;
static volatile int g_client_stop = 0; 
static uint32_t g_seq = 0;

static volatile sess_t   g_sess = {0};
static int g_channel_open = -1;

static trace_ret_t get_and_del_arg_from_hlist(const uint32_t seq, cb_arg_t **out)
{
    int ret = TRACE_FAILURE;
    uint32_t i = SESS_HASH_INDEX(seq);
    cb_arg_t *pos, *cb_arg = NULL;
    struct hlist_node *n;
    ret = pthread_mutex_lock((pthread_mutex_t *)&g_sess.hsess_mutex[i]);
    if (ret != 0) {
        SYSLOG_ERR("Thread locked session:[%u] is failure:%d", i, ret);
        goto END;
    }
    hlist_for_each_entry_safe(pos, n, &g_sess.hess[i], node) {
        if (pos->seq != seq) {
            continue;
        }
        hlist_del(&pos->node);  
        cb_arg = pos;
        
        g_sess.count[i]--;
        SYSLOG_DEBUG("Find cb by seq id:%u", seq);
        break;
    }
    ret = pthread_mutex_unlock((pthread_mutex_t *)&g_sess.hsess_mutex[i]);
    if (ret != 0) {
        SYSLOG_ERR("Thread unlocked session:[%u] is failure:%d", i, ret);
        goto END;
    }

    *out = cb_arg;
    ret = TRACE_SUCCESS;
END:
    return ret;
}

static void *cb_thread(void *arg)
{
    ssize_t size;
    trace_reply_t msg;

    return NULL;    // todo
    
    while(1) {
        if (g_client_stop) {
            SYSLOG_INFO("Callback thread is stopping");
            break;
        }

        // todo 需要考虑长期没收到消息，hash怎么清除数据
//        size = msgrcv(g_client_msgid, &msg, sizeof(msg.mtext), g_pid, MSG_NOERROR | IPC_NOWAIT);
        if ((size == -1)/* || (size != sizeof(msg.mtext))*/) {
            if ((errno != ENOMSG)) {
                printf("msg queue receive is failure:%d\n", errno);
            }

           // usleep(SLEEP_THREAD_TIME);     //防止CPU占用过高，睡眠一会儿，让出CPU todo
            continue;
        }

        cb_arg_t *cb_arg = NULL;
        if (get_and_del_arg_from_hlist(msg.hdr.seq, &cb_arg) == TRACE_FAILURE) {
            SYSLOG_ERR("Get arg is failure");
            break;
        }

        if (cb_arg == NULL) {
            SYSLOG_INFO("The seq:[%u] is not found", msg.hdr.seq);
            continue;
        }

        if (cb_arg->cb != NULL) {
            SYSLOG_DEBUG("Execute callback of seq:[%u]", msg.hdr.seq);
            cb_arg->cb(msg.result, cb_arg->arg);
        } else {
            SYSLOG_DEBUG("The callback of seq:[%u] is not set", msg.hdr.seq);
        }
        free(cb_arg); 
    }

    SYSLOG_INFO("Callback thread is stopped");
    return NULL;
}


static trace_ret_t cfg_channel_send(const uint32_t seq, const trace_policy_t  *policy, const reply_op_t is_reply)
{
    char buf[TRACE_POLICY_BUF_SZ + 10];
    struct nlmsghdr *hdr = (struct nlmsghdr *)buf;

    hdr->nlmsg_len = NLMSG_HDRLEN;
    hdr->nlmsg_flags = NLM_F_REQUEST /* | NLM_F_ACK */;
	hdr->nlmsg_type = TRACE_CFG_POLICY;
    hdr->nlmsg_pid = getpid();

    trace_req_t req;
    req.hdr.ver = 1;
    req.hdr.seq = seq;
    memcpy(&req.policy, policy, sizeof(*policy));
    
    commnl_addattr_l(hdr, sizeof(buf), TRACE_MSG_POLICY_REQ, &req, sizeof(trace_req_t));	
	SYSLOG_DEBUG("Send msg len:%u, msg_flag:%u, msg_type:%u", hdr->nlmsg_len, hdr->nlmsg_flags, hdr->nlmsg_type);
    SYSLOG_DEBUG("Send hdr: is_reply:%d, seq:%u, ver:%u", req.hdr.is_reply, req.hdr.seq, req.hdr.ver);
    SYSLOG_DEBUG("Send policy:");
    SYSLOG_DEBUG("    src family:%u, src ip:%02x, sport:%u", 
            req.policy.src.family, req.policy.src.addr.ip4, req.policy.sport);    
    SYSLOG_DEBUG("    dst family:%u, dst ip:%02x, dport:%u", 
            req.policy.dst.family, req.policy.dst.addr.ip4, req.policy.dport); 
    SYSLOG_DEBUG("    protocol:%u, app_type:%u", req.policy.protocol, req.policy.app_type); 

	/*发送组装好的netlink消息*/
	if (commcfg_send(hdr) < 0) {
        SYSLOG_ERR("Message(seq:%u) which been sent is failure", seq);
        return TRACE_FAILURE;
	}
    SYSLOG_INFO("Message(seq:%u) which been sent is success", seq);

    return TRACE_SUCCESS;
}


static void cfg_channel_close()
{
    if (g_channel_open >= 0) {
        commcfgnl_close();
    }
}

trace_ret_t trace_client_init()
{
    int i = 0;

    trace_ret_t pm_ret = collect_hlist_init((struct hlist_head *)g_sess.hess, sizeof(g_sess.hess) / sizeof(struct hlist_head));
    if (pm_ret != TRACE_SUCCESS) {
        SYSLOG_ERR("hlist init is failure:%d", pm_ret);
        goto FAIL;
    }

    for (i = 0; i < sizeof(g_sess.hsess_mutex) / sizeof(pthread_mutex_t); i++) {
        int ret = pthread_mutex_init((pthread_mutex_t *)&g_sess.hsess_mutex[i], NULL);
        if (ret != 0) {
            SYSLOG_ERR("Initial thread:[%d] is failure:%d", i, ret);
            goto FAIL;
        }
    }

    g_channel_open = commcfgnl_open();
    if(g_channel_open < 0)
	{
		SYSLOG_ERR("pdelivnl_open fail:%d", g_channel_open);
		goto FAIL;
	}
    
    int ret = pthread_create(&g_client_thread, NULL, cb_thread, NULL);
    if (ret != 0) {      
        SYSLOG_ERR("Create the thread of callback is failure:%d", ret);
        goto FAIL;
    }
    
    return TRACE_SUCCESS;
FAIL:
    while (i > 0) {
        i--;
        pthread_mutex_destroy((pthread_mutex_t *)&g_sess.hsess_mutex[i]);
    }
    cfg_channel_close();
    
    return TRACE_FAILURE;
}

trace_ret_t trace_client_exit()
{
    g_client_stop = 1;
    pthread_join(g_client_thread, NULL);

    for (int i = 0; i < sizeof(g_sess.hsess_mutex) / sizeof(pthread_mutex_t); i++) {
        pthread_mutex_destroy((pthread_mutex_t *)&g_sess.hsess_mutex[i]);
    }

    cb_arg_t *pos;
    struct hlist_node *n;
    COLLECT_HLIST_CLEAR(pos, n, g_sess.hess, sizeof(g_sess.hess) / sizeof(struct hlist_head), node, free);

    cfg_channel_close();

    return TRACE_SUCCESS;
}

static cb_arg_t *add_to_list_and_get(const uint32_t i, const uint32_t seq, const struct sysinfo *info, async_cb cb, void *arg)
{
    cb_arg_t *cb_arg = NULL; 
    struct hlist_node *prev_last;
    async_cb tmp_cb = NULL;
    void *tmp_arg;
       
    prev_last = g_sess.last[i];
    if (g_sess.count[i] >= MAX_QUEUE_COUNT) {
        cb_arg_t *first = hlist_entry(g_sess.hess[i].first, cb_arg_t, node);
        if ((info->uptime - first->t) > MAX_QUEUE_TIMEOUT) {            
            hlist_del(&first->node);          

            tmp_cb = cb_arg->cb;
            tmp_arg = cb_arg->arg;
            
            // 复用cb_arg
            cb_arg = first;
        } else {            
            SYSLOG_WARN("Hash table:[%u] queue is full", i);
            goto FAIL;
        }
    }

    if (cb_arg == NULL) {
        cb_arg = (cb_arg_t *)malloc(sizeof(*cb_arg));
        if (cb_arg == NULL) {
            SYSLOG_ERR("Allocateing arg of callback is failure:%d\n", errno);
            goto FAIL;
        }
    }
    cb_arg->seq = seq;
    cb_arg->arg = arg;
    cb_arg->cb = cb;
    INIT_HLIST_NODE(&cb_arg->node);
    
    cb_arg->t = info->uptime;  

    // 新节点加的链表后面
    if (hlist_empty((struct hlist_head *)&g_sess.hess[i])) {
        hlist_add_head(&cb_arg->node, (struct hlist_head *)&g_sess.hess[i]);
    } else {
        hlist_add_behind(&cb_arg->node, g_sess.last[i]);
    }
    g_sess.last[i] = &cb_arg->node;
    g_sess.count[i]++;
   
    return cb_arg;
FAIL:
    if (cb_arg == NULL) {
        free(cb_arg);
    }
    
    return NULL;
}

static trace_ret_t __trace_async_exec(const trace_policy_t  *in, 
                            async_cb cb, void *arg, uint32_t *seq_out)
{
    trace_ret_t ret = TRACE_FAILURE;
    struct hlist_node *prev_last;
    async_cb tmp_cb = NULL;
    void *tmp_arg;
    struct sysinfo info;
    uint32_t seq, i;
    reply_op_t is_reply = REPLY_OP_NO_NEED;

    seq = __sync_add_and_fetch(&g_seq, 1);
    SYSLOG_DEBUG("The seq of the message is %u", seq);

    cb_arg_t *cb_arg = NULL; 
    int ret_thread;
    if (cb != NULL) {
        i = SESS_HASH_INDEX(seq);
        if (sysinfo(&info) != 0) {        
            SYSLOG_ERR("Get current boot time is failure:%d\n", errno);
            goto END;
        }

        ret_thread = pthread_mutex_lock((pthread_mutex_t *)&g_sess.hsess_mutex[i]);
        if (ret_thread != 0) {
            SYSLOG_ERR("Async Interface:locked session:[%u] is failure:%d", i, errno);
            goto END;
        }
    
        prev_last = g_sess.last[i];
        cb_arg = add_to_list_and_get(i, seq, &info, cb, arg);
        ret_thread = pthread_mutex_unlock((pthread_mutex_t *)&g_sess.hsess_mutex[i]);
        if (ret_thread != 0) {
            SYSLOG_ERR("Async Interface:unlocked session:[%u] is failure:%d", i, errno);
            goto END;
        }
        if (cb_arg == NULL) {
            SYSLOG_ERR("cbarg which is been added is failure");
            goto END;
        } 

        is_reply = REPLY_OP_NEED;
    }

    ret = cfg_channel_send(seq, in, is_reply);
    if (ret != TRACE_SUCCESS) {
        SYSLOG_ERR("Sending cfg is failure");
        goto END;
    }

    if (seq_out != NULL) {
        *seq_out = seq;
    }

    if (cb != NULL) {
        ret = TRACE_PENDING;
    }
END:
    if (cb != NULL) {
        if ((ret == TRACE_FAILURE)
            && (cb_arg != NULL)) {
            ret_thread = pthread_mutex_lock((pthread_mutex_t *)&g_sess.hsess_mutex[i]);
            if (ret_thread != 0) {
                SYSLOG_ERR("Free data:locked session:[%u] is failure:%d", i, errno);
                goto END2;
            }
            hlist_del(&cb_arg->node);
            free(cb_arg);

            g_sess.last[i] = prev_last;
            g_sess.count[i]--;
            ret_thread = pthread_mutex_unlock((pthread_mutex_t *)&g_sess.hsess_mutex[i]);        
            if (ret_thread != 0) {
            SYSLOG_ERR("Free data:unlocked session:[%u] is failure:%d", i, errno);
            goto END;
        }
        }
    }

END2:
    if (tmp_cb != NULL) {
        tmp_cb(TRACE_FAILURE, tmp_arg);
    }

    return ret;
}

trace_ret_t trace_async_exec(const trace_policy_t  *in, 
                        async_cb cb, void *arg)
{
    return __trace_async_exec(in, cb, arg, NULL);
}


static void sync_exec_cb(trace_ret_t ret, void *arg)
{
    sync_arg_t *a = (sync_arg_t *)arg;

    SYSLOG_DEBUG("sync cb");
    a->ret = ret;
    if (sem_post(&a->sem) != 0) {
        SYSLOG_ERR("Set semaphore is failure:%d", errno);
    }
}

trace_ret_t trace_sync_exec(const trace_policy_t  *in)
{
    trace_ret_t ret = TRACE_FAILURE;
    sync_arg_t *arg = (sync_arg_t *)malloc(sizeof(*arg));
    if (arg == NULL) {
        SYSLOG_ERR("Allocate sync arg is failure:%d", errno);
        return TRACE_FAILURE;
    }
    if (sem_init(&arg->sem, 0, 0) != 0) {
        SYSLOG_ERR("Init sem is failure:%d", errno);
        goto END1;
    }

    uint32_t seq;
    if (__trace_async_exec(in, sync_exec_cb, arg, &seq) == TRACE_FAILURE) {
        SYSLOG_ERR("Exec policy is failure");
        goto END2;
    }
    
    struct timespec timeout;
    if (clock_gettime(CLOCK_REALTIME, &timeout) == -1) {
        SYSLOG_ERR("Get current time is failure:%d", errno);
        goto END2;
    }

    timeout.tv_sec += EXEC_SYNC_WAIT_TIMEOUT;
    if (sem_timedwait(&arg->sem, &timeout) == -1) {
        cb_arg_t *cb_arg = NULL;
        ret = get_and_del_arg_from_hlist(seq, &cb_arg);
        if (ret == TRACE_SUCCESS) {
            if (cb_arg != NULL) {
                free(cb_arg);
                arg->ret = TRACE_FAILURE;
                SYSLOG_ERR("Wait exec result is failure:%d", errno);
            }
        } else {
            SYSLOG_ERR("Get arg is failure");
            goto END2;
        } 
    }    

    SYSLOG_INFO("Sync exec is completely, respone result:%u", arg->ret);
    ret = arg->ret;
END2:
    sem_destroy(&arg->sem);
END1:
    free(arg);
    return ret; 
}

trace_ret_t trace_exec_no_reply(const trace_policy_t  *in)
{
    return __trace_async_exec(in, NULL, NULL, NULL);
}