secgateway/Platform/user/trace/trace-api/trace_api.c

#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 <arpa/inet.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)

#define NETLINK_GROUP_TRACE_ID      0

#define EXEC_CR_SK_WAIT_TIMEOUT     2


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 sem_t g_cr_sk_sem;


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 trace_ret_t msg_handle(const trace_reply_t *msg)
{
    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");
        return TRACE_FAILURE;
    }

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

    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); 

    return TRACE_SUCCESS;
}

static int trace_recv_handle(struct pdelivnl_ctrl_data *ctrl,
				    struct nlmsghdr *n, void *arg) 
{
    trace_reply_t *reply;
    trace_ret_t ret;
    
    SYSLOG_INFO("Trace receives reply message, msg_type:%u", n->nlmsg_type);
    switch (n->nlmsg_type) {
        case TRACE_CFG_POLICY_REPLY:
            reply = (trace_reply_t *)NLMSG_DATA(n);
            if (sizeof(*reply) < (n->nlmsg_len - NLMSG_HDRLEN)) {
                SYSLOG_WARN("The length of the reply message is required to be %ld, but fact length is %u", 
                    sizeof(*reply), (n->nlmsg_len - NLMSG_HDRLEN));
                break;
            }
            ret = msg_handle(reply);
            if (ret != TRACE_SUCCESS) {
                SYSLOG_ERR("Processing message is fail");
            }
           
            break;
        default:
            SYSLOG_WARN("Unknown type:%u of the message is received from netlink", n->nlmsg_type);
            break;
    }

    SYSLOG_INFO("Reply message of trace finalizes, msg_type:%u", n->nlmsg_type);

    return 0;
}


static void *cb_thread(void *arg)
{
    SYSLOG_INFO("Callback thread is started");
	
    g_channel_open = pdelivnl_open(NETLINK_GROUP_TRACE_ID);
    if(g_channel_open < 0)
    {
        SYSLOG_ERR("pdelivnl_open fail:%d", g_channel_open);
        return NULL;
    }

    if (sem_post(&g_cr_sk_sem) != 0) {
        SYSLOG_ERR("Set semaphore of g_cr_sk_sem is failure:%d", errno);
    }

    pdelivnl_listen(NETLINK_GROUP_TRACE_ID, trace_recv_handle, NULL);
    
    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_REQ;
 //   hdr->nlmsg_pid = getpid();

    trace_req_t req;
    req.hdr.ver = 1;
    req.hdr.seq = seq;
    req.hdr.is_reply = is_reply;
    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 netlink-message-pid:%u ", hdr->nlmsg_pid);
    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 (%s), sport:%u (host order %u)", 
            req.policy.src.family, req.policy.src.addr.ip4.s_addr, inet_ntoa(req.policy.src.addr.ip4),
            req.policy.sport, ntohs(req.policy.sport));    
    SYSLOG_DEBUG("    dst family:%u, dst ip:%02x (%s), dport:%u (host order %u)", 
            req.policy.dst.family, req.policy.dst.addr.ip4.s_addr, inet_ntoa(req.policy.dst.addr.ip4),
            req.policy.dport, ntohs(req.policy.dport)); 
    SYSLOG_DEBUG("    protocol:%u, app_type:%u, base_type:%u", req.policy.protocol, req.policy.app_type, req.policy.base_type); 

	/*发送组装好的netlink消息*/
    //if (commcfg_send(hdr) < 0) {
	if (pdelivnl_send(NETLINK_GROUP_TRACE_ID, 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) {
         pdelivnl_close(NETLINK_GROUP_TRACE_ID);
    }
}

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();
    /* netlink通信机制决定同一个线程里不能同时启用两个pdeliv socket，
     * 鉴于trace库的调用者（如DPI）可能原本已启用了一个pdeliv socket，
     * 因此，将trace库里的pdeliv socket创建移至trace的子线程中 */
    #if 0 
    g_channel_open = pdelivnl_open(NETLINK_GROUP_TRACE_ID);
    if(g_channel_open < 0)
	{
		SYSLOG_ERR("pdelivnl_open fail:%d", g_channel_open);
		goto FAIL;
	}
    #endif

    if (sem_init(&g_cr_sk_sem, 0, 0) != 0) {
        SYSLOG_ERR("Init g_cr_sk_sem is failure:%d", errno);
        goto FAIL;
    }

    struct timespec timeout;
    if (clock_gettime(CLOCK_REALTIME, &timeout) == -1) {
        SYSLOG_ERR("Get current time is failure:%d", errno);
        goto FAIL;
    }

    timeout.tv_sec += EXEC_CR_SK_WAIT_TIMEOUT;
    
    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;
    }

    /* 如果sem 信号量值>0，则sem_timedwait 立即返回；
     * 如果sem 信号量值≤0，则 sem_timedwait 阻塞等待 TIMEOUT秒后再返回。*/
    ret = sem_timedwait(&g_cr_sk_sem, &timeout);
    SYSLOG_DEBUG("sem_timedwait for g_cr_sk_sem, return:%d", ret);

    sem_destroy(&g_cr_sk_sem);
    return TRACE_SUCCESS;
FAIL:
    while (i > 0) {
        i--;
        pthread_mutex_destroy((pthread_mutex_t *)&g_sess.hsess_mutex[i]);
    }
    cfg_channel_close();

    sem_destroy(&g_cr_sk_sem);
    return TRACE_FAILURE;
}

trace_ret_t trace_client_exit()
{
    //g_client_stop = 1;
    cfg_channel_close();    // 先关闭，判断SOCKET异常来退出线程
    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);

    

    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;
        SYSLOG_DEBUG("Need reply:%u from kernel", is_reply);
    }

    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);
}