path: root/src/mrl_io.cpp

#include <arpa/inet.h>
#include <sys/types.h>
#include <assert.h>
#include<unistd.h>
#include <stdio.h>

#include "stream.h"
#include "stream_inc/stream_control.h"
#include "mrtunnat.h"

#include "cJSON.h"
#include "mrl_io.h"
#include "mrl_main.h"
#include "mrl_utils.h"
#include "mrl_redis.h"
#include "mrl_stat.h"

extern struct global_stat_t global_stat;

extern struct mrl_global_instance mrl_instance;

bool mrl_get_link_identity(const char* back_ip, uint64_t* link_identity);
void mrl_detect_ip(const char* src_ip, int nominee_type);


void mrl_socket_init()
{
    int i;
    int thread_num = 0;
    int len = sizeof(thread_num);
    sapp_get_platform_opt(SPO_THREAD_COUNT,(void *) &thread_num, &len);
    mrl_instance.mrl_snd_fd = (int *) calloc (1,sizeof(int)*thread_num);
    global_stat.malloc_memory += sizeof(int)*thread_num;
    
    for(i = 0; i< thread_num;i++)
    {
        mrl_instance.mrl_snd_fd[i] = socket(AF_INET, SOCK_DGRAM, 0);   
    }
       
    //init destnation sock addr
    mrl_instance.mgw_addr.sin_family = AF_INET;
    inet_pton(AF_INET, mrl_instance.mrl_cfg.mgw_ip, &(mrl_instance.mgw_addr.sin_addr));
    mrl_instance.mgw_addr.sin_port = htons(mrl_instance.mrl_cfg.mgw_port);

    //init recv fd
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = htons(mrl_instance.mrl_cfg.mrl_port);
    mrl_instance.mrl_rcv_fd = socket(AF_INET, SOCK_DGRAM, 0);
    bind(mrl_instance.mrl_rcv_fd , (struct sockaddr *)(&addr), sizeof(addr));
   
    
}

void mrl_socket_close()
{
    int i;
    int thread_num = 0;
    int len = sizeof(thread_num);
    sapp_get_platform_opt(SPO_THREAD_COUNT,(void *) &thread_num, &len);
    for(i = 0; i< thread_num;i++)
    {
      close(mrl_instance.mrl_snd_fd[i]);
    }

    free(mrl_instance.mrl_snd_fd);
    global_stat.free_memory += sizeof(int)*thread_num;
    
    close(mrl_instance.mrl_rcv_fd);
}

ssize_t mrl_sock_send(int fd, const char *pkt, ssize_t pkt_len, struct sockaddr_in *dest)
{
    ssize_t send_len = 0;
    while(1)
    {
        send_len = sendto(fd, pkt,pkt_len,0,(struct sockaddr *)dest,sizeof(struct sockaddr_in));
        if(send_len < 0)
        {
            if(errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
            {
                continue;
            }
            else
            {
                assert(0);
            }
            
        }
        else
        {
            break;
        }
    }
    return send_len;
}

ssize_t mrl_sock_recv(int fd, char *recv_buff, size_t buff_len)
{
    struct sockaddr_in from;
    socklen_t from_len = sizeof(from);
    ssize_t recv_len = 0;
    while(1)
    {
         recv_len = recvfrom(fd, recv_buff, buff_len, 0, (struct sockaddr *) &from, & from_len);
         if(recv_len < 0)
         {
            if(errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
            {
                continue;
            }
            else
            {
                assert(0);
            }
            
         }
         else
         {
            break;
         }
    }
    return recv_len ;
}


void mrl_send_to_gdev(int thread_seq, uint64_t link_identity, const char *payload, size_t payload_len)
{ 
    SAPP_TLV_T option[32];
    int option_num = 0;
 
    option[option_num].type = SAPP_SEND_OPT_VIRTUAL_LINK_ID;
    option[option_num].length = 8;
    option[option_num].long_value = link_identity;
    option_num++;///////
    
    int ret = 0;
    ret = MESA_sendpacket_iplayer_options(thread_seq,  
                         payload,
                         payload_len,
                         0,
                         option,
                         option_num);
    assert(ret >=0);
}

int mrl_send_to_mgw(void *raw_pkt, int thread_seq)
{
    const char *pkt = (const char *)raw_pkt;
    struct iphdr *ip_hdr = (struct iphdr *)pkt;
    ssize_t pkt_len = ntohs(ip_hdr->tot_len);
    ssize_t send_len = 0;
    send_len = mrl_sock_send(mrl_instance.mrl_snd_fd[thread_seq],pkt,pkt_len,&(mrl_instance.mgw_addr));
    assert(send_len == pkt_len);
    return send_len;
    
}

void *mrl_recv_from_mgw(void *arg)
{
    char rcv_buff[MRL_BUFF_LEN];
    ssize_t recv_len = 0;
    int thread_seq = 0;
    int temp_len = 0;
    struct mrl_tuple tuple;
    uint64_t link_identity = 0;
    while(1)
    {
        if(mrl_instance.stop_flag == 1)
        {
            break;
        }
        memset(rcv_buff,0,MRL_BUFF_LEN);
        recv_len = mrl_sock_recv(mrl_instance.mrl_rcv_fd,rcv_buff, MRL_BUFF_LEN);
        global_stat.recv_mgw_pkts ++;
        memset(&tuple,0,sizeof(struct mrl_tuple));
        mrl_get_pkt_tuple(rcv_buff,&tuple);
        MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_recv_from_mgw","recv from mgw pkt info:"
                                "[sip:%s,sport:%hu,dip:%s,dport:%hu,protocol:%hu,pkt_len:%d]",
                                tuple.sip,tuple.sport,tuple.dip,tuple.dport,recv_len);
                
        if(mrl_get_link_identity(tuple.sip,&link_identity))
       	{
       		temp_len = sizeof(thread_seq);
        	sapp_get_platform_opt(SPO_INDEPENDENT_THREAD_ID, &thread_seq, &temp_len);
        	mrl_send_to_gdev(thread_seq,link_identity,rcv_buff, recv_len);
        }
       
        
    }
    return NULL;
}


size_t mrl_build_ip_hdr(char *ip_pkt, uint32_t src_ip, uint32_t dst_ip, size_t udp_len)
{

    size_t ip_len = udp_len + MRL_IP_HDR_LEN;
    struct iphdr *ip_hdr =(struct iphdr *)ip_pkt;
    ip_hdr->ihl = 0x5;
    ip_hdr->version=0x4;
    ip_hdr->tos = 0x00;
    ip_hdr->tot_len = htons(ip_len);
    ip_hdr->id = 0x325f;
    ip_hdr->frag_off =0x0000;
    ip_hdr->ttl = 0x80;
    ip_hdr->protocol =0x11;
    ip_hdr->check = 0x0000;
    ip_hdr->saddr = src_ip;
    ip_hdr->daddr = dst_ip;
    ip_hdr->check = mrl_get_checksum((uint16_t *)ip_pkt,MRL_IP_HDR_LEN);
    return ip_len;
}

size_t mrl_build_udp_hdr(char *udp_pkt,size_t payload_len)
{
    struct udphdr *udp_hdr = (struct udphdr *)udp_pkt;
    udp_hdr->source = htons(mrl_instance.mrl_cfg.local_port);
    udp_hdr->dest = htons(mrl_instance.mrl_cfg.dest_port);
    udp_hdr->len = htons(payload_len + MRL_UDP_HDR_LEN);
    udp_hdr->check = 0;
    return payload_len + MRL_UDP_HDR_LEN;
}

size_t mrl_build_udp_payload(char *payload,uint64_t link_identity, int nominee_type)
{
    char content[MRL_MAX_PAYLOAD_LEN];
    memset(content,0,MRL_MAX_PAYLOAD_LEN);
    snprintf(content,MRL_MAX_PAYLOAD_LEN,"nominee_type:%d,link_identity:%lu",nominee_type, link_identity);
    size_t payload_len = strlen(content) +1;
    strncpy(payload,content,payload_len);
	MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_build_udp_payload",
							"cur detect udp payload is %s",content);
    return payload_len;
}


MESA_htable_handle mrl_htable_init(void * fn_data_free_cb)
{
	unsigned int opt_int;
    MESA_htable_handle htable = MESA_htable_born();
    assert(htable != NULL);
    
    opt_int = 1;
    MESA_htable_set_opt(htable, MHO_THREAD_SAFE, &opt_int, sizeof(int));

    opt_int = 1;
    MESA_htable_set_opt(htable, MHO_MUTEX_NUM, &opt_int, sizeof(int));
    
    opt_int = 0;
    MESA_htable_set_opt(htable, MHO_EXPIRE_TIME, &opt_int, sizeof(int));

    opt_int = 0;
    MESA_htable_set_opt(htable, MHO_AUTO_UPDATE_TIME, &opt_int, sizeof(int));

    opt_int = 0;
    MESA_htable_set_opt(htable, MHO_SCREEN_PRINT_CTRL, &opt_int, sizeof(int));

    opt_int = mrl_instance.mrl_cfg.ht_slot_size;
    MESA_htable_set_opt(htable, MHO_HASH_SLOT_SIZE, &opt_int, sizeof(int));
    
    opt_int = mrl_instance.mrl_cfg.ht_max_element_num;
    MESA_htable_set_opt(htable, MHO_HASH_MAX_ELEMENT_NUM, &opt_int, sizeof(int));

    opt_int = mrl_instance.mrl_cfg.ht_mutex_num;
    MESA_htable_set_opt(htable, MHO_MUTEX_NUM, &opt_int, sizeof(int));

    MESA_htable_set_opt(htable, MHO_CBFUN_DATA_FREE, fn_data_free_cb, sizeof(fn_data_free_cb));

    int ret = MESA_htable_mature(htable);
    if (ret < 0)
    {
       MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_FATAL,"mrl_init_htable","MESA_htable_mature func error!");
        assert(0);
    }
   return htable;
}

void ht_nominee_iterate_cb(const uchar* key, uint size, void* data, void* user)
{
    struct mrl_nominee* nominee = (struct mrl_nominee*)data;
	if(nominee!= NULL)
	{
		MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_nominee_iterate_cb","pending detect nominee ip is %s,nominee_type is %d",nominee->ip_addr,nominee->nominee_type);
		switch(nominee->nominee_type)
		{
			case MRL_SNAT_NOMINEE_TYPE:
				if(MESA_htable_search_cb(mrl_instance.ht_snat_candidate, (const unsigned char *)nominee->ip_addr, MRL_STR_IP_LEN,NULL,NULL,NULL) == NULL)
				{	
					MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_nominee_iterate_cb","start to detect snat nominee ip %s",nominee->ip_addr);
					mrl_detect_ip(nominee->ip_addr,nominee->nominee_type);
				}
				else
				{					
					MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_nominee_iterate_cb","cur snat nominee ip %s has became snat candidate ip.",nominee->ip_addr);
				}
				break;
			case MRL_DNAT_NOMINEE_TYPE:
				if(MESA_htable_search_cb(mrl_instance.ht_dnat_candidate, (const unsigned char *)nominee->ip_addr, MRL_STR_IP_LEN,NULL,NULL,NULL) == NULL)
				{
					MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_nominee_iterate_cb","start to detect dnat nominee(dnat_policy) ip %s",nominee->ip_addr);
					mrl_detect_ip(nominee->ip_addr,nominee->nominee_type);
				}
				else
				{	  
					MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_nominee_iterate_cb","cur dnat nominee(dnat_policy) ip has became dnat candidate ip %s",nominee->ip_addr);
				}
				break;
			default:
				MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_FATAL,"ht_nominee_iterate_cb","cur nominee ip  type is unknow,type is %d",nominee->nominee_type);
				break;
		}
    }
    
}


void ht_link_identity_iterate_cb(const uchar* key, uint size, void* data, void* user)
{
    uint32_t dip =0, sip = 0;
    size_t udp_payload_len = 0;
    size_t udp_len = 0;
    size_t ip_len = 0;
    int thread_seq = 0;
    int temp_len = 0;
    char ip_pkt[MRL_IP_PKT_LEN];
    memset(ip_pkt,0,MRL_IP_PKT_LEN);
    uint64_t* identity_value = (uint64_t* )data;
	if(identity_value != NULL)
	{
		const uint64_t link_identity = *identity_value;
    	struct user_data *mydata = (struct user_data* )user;
    	udp_payload_len = mrl_build_udp_payload(ip_pkt +MRL_UDP_HDR_LEN + MRL_IP_HDR_LEN,link_identity,mydata->nominee_type);
    	udp_len = mrl_build_udp_hdr(ip_pkt + MRL_IP_HDR_LEN, udp_payload_len);
    	inet_pton(AF_INET,mydata->detect_ip,&sip);
    	inet_pton(AF_INET,mrl_instance.mrl_cfg.dest_ip,&dip);
    	ip_len = mrl_build_ip_hdr(ip_pkt,sip,dip,udp_len);
    	sapp_get_platform_opt(SPO_INDEPENDENT_THREAD_ID, &thread_seq, &temp_len);
    	mrl_send_to_gdev(thread_seq,link_identity,ip_pkt,ip_len);
		MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_link_identity_iterate_cb",
							"cur detect ip %s detect info[nominee_type:%d,link_identity:%lu]",
							mydata->detect_ip,mydata->nominee_type,link_identity);
    	global_stat.send_detect_pkts ++;
	
}
    
}
/*
long ht_dnat_policy_search_cb(void *data, const uchar *key, uint size, void *user_arg)
{
    struct mrl_dnat_policy* dnat_policy = (struct mrl_dnat_policy* )data;
    struct mrl_dnat_policy* user_dnat_policy = (struct mrl_dnat_policy* )user_arg;
    
    if(dnat_policy != NULL)//�û��������ö˿ڣ�
    {
         MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"dnat_policy_update_cb","get ht_dnat_policy value:"
                "[config_id:%d,original_addr_type:%d,original_dest_ip:%s,original_dest_port:%s,original_protocol:%s,translated_dest_ip:%s,"
                "translated_dest_port:%s,do_log:%d,action:%d,addr_type:%d,service:%d,is_valid:%d,effective_range:%s,op_time:%s]", 
                dnat_policy->config_id,dnat_policy->original_addr_type,dnat_policy->original_dest_ip,dnat_policy->original_dest_port,dnat_policy->original_protocol,
                dnat_policy->translated_dest_ip, dnat_policy->translated_dest_port,dnat_policy->do_log,dnat_policy->action,
                dnat_policy->service,dnat_policy->is_valid,dnat_policy->effective_range, dnat_policy->op_time);
        memcpy(user_dnat_policy,dnat_policy,sizeof(struct mrl_dnat_policy));
        return 1;
    }
    else
    {
        return 0;
    }
}*/
long ht_nominee_search_cb(void *data, const uchar *key, uint size, void *user_arg)
{
    struct mrl_nominee* nominee = (struct mrl_nominee* )data;
    struct mrl_nominee *user_nominee = (struct mrl_nominee*) user_arg;
    
    if(nominee != NULL)//�û��������ö˿ڣ�
    {
        MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_nominee_search_cb","get nominee value:"
                "[config_id:%d,addr_pool_id:%d,addr_type:%d,ip_addr:%s,is_valid:%d,effective_range:%s,op_time:%s]", 
                nominee->config_id, nominee->addr_pool_id,nominee->addr_type, 
                nominee->ip_addr, nominee->is_valid,nominee->effective_range,nominee->op_time);
        memcpy(user_nominee,nominee,sizeof(struct mrl_nominee));
        return 1;
    }
    else
    {
    	MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_nominee_search_cb","cur ht_nominee is NULL!");
        return 0;
    }
  
}


bool mrl_dnat_key_search(struct streaminfo *mystream, struct mrl_dnat_policy_key *dnat_policy_key)
{
    if(MESA_htable_search_cb(mrl_instance.ht_dnat_policy, (const unsigned char *)dnat_policy_key, sizeof(struct mrl_dnat_policy_key),NULL,NULL,NULL) != NULL)
    {
        return  true;
    }
    else
    {
        return false;
    }
}


int mrl_htable_delete(MESA_htable_handle htable,const unsigned char* key,unsigned int key_len)
{
    int ret = 0;
    if(MESA_htable_search_cb(htable, key, key_len,NULL,NULL,NULL) != NULL)
    {
        ret = MESA_htable_del(htable, key, key_len, NULL);
        if(ret < 0)
        {
            return ret;
        }
        else
        {
            return MRL_HTABLE_DEL_SUCCESS;
        }
    }
    else
    {
             MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_htable_delete","the key is not exsit in ht_nominee.");

             return MRL_HTABLE_DEL_NULL;
    }
}


int mrl_htable_add(MESA_htable_handle htable,const unsigned char* key,unsigned int key_len,const void* value)
{
    int ret =0;
    if(MESA_htable_search_cb(htable, key, key_len,NULL,NULL,NULL) == NULL)
    {
        ret = MESA_htable_add(htable, key, key_len, value);
        if(ret < 0)
        {
           return ret;
        }
        else
        {
            return MRL_HTABLE_ADD_SUCCESS;
      
        
}
     }
     else
     {
         return MRL_HTABLE_ADD_DUPLICATE;
     }
}
void ht_link_identity_free_cb(void * data)
{
    uint64_t* link_identity = (uint64_t* )data;
    if(link_identity != NULL)
    {
         MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_link_identity_free_cb",
                                "the link identity item:[link_identity:%d]",*link_identity);
        link_identity = NULL;
        global_stat.free_memory += sizeof(uint64_t);
    }
        
}

void ht_nominee_free_cb(void * data)
{
    struct mrl_nominee* nominee = (struct mrl_nominee* )data;
    if(nominee != NULL)
    {
         MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_snat_nominee_free_cb","the nominee item:"
                "[config_id:%d,addr_pool_id:%d,addr_type:%d,ip_addr:%s,is_valid:%d,effective_range:%s,op_time:%s] in ht_nominee is free", 
                nominee->config_id, nominee->addr_pool_id,nominee->addr_type, 
                nominee->ip_addr, nominee->is_valid,nominee->effective_range,nominee->op_time);
        free(nominee);
        nominee = NULL;
        global_stat.free_memory += sizeof(struct mrl_nominee);
    }
        
}

void ht_snat_candidate_free_cb(void * data)
{
    struct mrl_snat_candidate* snat_candidate = (struct mrl_snat_candidate* )data;
    if(snat_candidate != NULL)
    {
         MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_candidate_free_cb","the candidate item:"
                "[config_id:%d,addr_pool_id:%d,addr_type:%d,ip_addr:%s,location:%d,is_valid:%d,effective_range:%s,op_time:%s] in ht_nominee is free", 
                snat_candidate->config_id, snat_candidate->addr_pool_id,snat_candidate->addr_type, 
                snat_candidate->ip_addr, snat_candidate->location,snat_candidate->is_valid,
                snat_candidate->effective_range,snat_candidate->op_time);
        free(snat_candidate);
        snat_candidate = NULL;
        global_stat.free_memory += sizeof(struct mrl_snat_candidate);
    }
        
}

void ht_dnat_policy_free_cb(void * data)
{
    struct mrl_dnat_policy *dnat_policy = (struct mrl_dnat_policy *)data;
    if(dnat_policy != NULL)
    {
        MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_dnat_policy_free_cb","the dnat_policy item:"
                "[config_id:%d,original_addr_type:%d,original_dest_ip:%s,original_dest_port:%s,original_protocol:%s,translated_user_type:%s,translated_user_id:%s,"
                "translated_dest_port:%s,do_log:%d,action:%d,addr_type:%d,service:%d,is_valid:%d,effective_range:%s,op_time:%s] in ht_dnat_policy is free", 
                dnat_policy->config_id,dnat_policy->original_addr_type,dnat_policy->original_dest_ip,dnat_policy->original_dest_port,dnat_policy->original_protocol,
                dnat_policy->translated_user_type,dnat_policy->translated_user_id, dnat_policy->translated_dest_port,dnat_policy->do_log,dnat_policy->action,
                dnat_policy->service,dnat_policy->is_valid,dnat_policy->effective_range, dnat_policy->op_time);
        free(dnat_policy);
        dnat_policy = NULL;
        global_stat.free_memory += sizeof(struct mrl_dnat_policy);
    }
        
}



void ht_dnat_candidate_free_cb(void * data)
{
    struct mrl_dnat_candidate* dnat_candidate = (struct mrl_dnat_candidate* )data;
    if(dnat_candidate != NULL)
    {
         MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_dnat_candidate_free_cb","the dnat_candidate item:"
                "[config_id:%d,addr_type:%d,ip_addr:%s,location:%d,is_valid:%d,effective_range:%s,op_time:%s] in ht_dnat_candidate is free", 
                dnat_candidate->config_id,dnat_candidate->addr_type, dnat_candidate->ip_addr,dnat_candidate->location,
                dnat_candidate->is_valid,dnat_candidate->effective_range,dnat_candidate->op_time);
        free(dnat_candidate);
        dnat_candidate = NULL;
        global_stat.free_memory += sizeof(struct mrl_dnat_candidate);
    }
        
}

void ht_vxlan_info_free_cb(void * data)
{
    struct mrl_vxlan_info* vxlan_info = (struct mrl_vxlan_info* )data;
    if(vxlan_info != NULL)
    {
        MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_vxlan_info_free_cb","the vxlan_info item:"
                                "[config_id:%d,back_ip:%s,back_type:%d,link_identity:%lu,"
                                "is_vaild:%d,effective_range:%s,op_time:%s] in ht_vxlan_info is free",
                                vxlan_info->config_id, vxlan_info->back_ip, vxlan_info->back_type,vxlan_info->link_identity,
                                vxlan_info->is_valid, vxlan_info->effective_range, vxlan_info->op_time);
        free(vxlan_info);
        vxlan_info = NULL;
        global_stat.free_memory += sizeof(struct mrl_vxlan_info);
    }
        
}

void ht_mrl_ip_info_free_cb(void * data)
{
    struct mrl_mrl_ip *mrl_ip = (struct mrl_mrl_ip *)data;
    if(mrl_ip != NULL)
    {
        MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"ht_vxlan_info_free_cb","the vxlan_info item:"
                                "[config_id:%d,back_ip:%s,back_type:%d,mrl_ip:%s,is_valid:%d,effective_range:%s,op_time:%s] in ht_mrl_ip_info is free", 
                                mrl_ip->config_id, mrl_ip->back_ip, mrl_ip->back_type,mrl_ip->mrl_ip,
                                mrl_ip->is_valid,mrl_ip->effective_range,mrl_ip->op_time);
        free(mrl_ip);
        mrl_ip = NULL;
        global_stat.free_memory += sizeof(struct mrl_mrl_ip);
    }
        
}

//�ж��Ƿ���dnat���ݰ�
bool mrl_dnat_pkt_identify(struct streaminfo *mystream, struct mrl_tuple *tuple)
{
    bool ret = false;
    MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_dnat_pkt_identify","sapp recv stream:"
                            "[sip:%s,sport:%hu,dip:%s,dport:%hu,portocol:%hu] start dnat identify",tuple->sip,tuple->sport,tuple->dip,tuple->dport,tuple->protocol);
    struct mrl_dnat_policy_key dnat_policy_key;
    memset(&dnat_policy_key,0,sizeof(struct mrl_dnat_policy_key));
    
    strncpy(dnat_policy_key.original_ip,(const char*)tuple->dip,MRL_STR_IP_LEN);
    dnat_policy_key.original_port = tuple->dport;
    
    if(mrl_dnat_key_search(mystream,&dnat_policy_key))
    {
        MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_dnat_pkt_identify","sapp recv stream info:"
                            "[sip:%s,sport:%hu,dip:%s,dport:%hu,portocol:%hu] is dnat pkt",tuple->sip,tuple->sport,tuple->dip,tuple->dport,tuple->protocol);
        ret = true;
    }
    else
    {
        dnat_policy_key.original_port = 0;
        if(mrl_dnat_key_search(mystream,&dnat_policy_key))
        {
            MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_dnat_pkt_identify","sapp recv stream info:"
                            "[sip:%s,sport:%hu,dip:%s,dport:%hu,portocol:%hu] is dnat pkt",tuple->sip,tuple->sport,tuple->dip,tuple->dport,tuple->protocol);
            ret = true;
        }
        else
        {
            MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_dnat_pkt_identify","sapp recv stream info:"
                            "[sip:%s,sport:%hu,dip:%s,dport:%hu,portocol:%hu] is not dnat pkt",tuple->sip,tuple->sport,tuple->dip,tuple->dport,tuple->protocol);
            ret = false;
        }
    }
    return ret;
    
}


//�ж��Ƿ���snat���ݰ�
bool mrl_snat_pkt_identify(struct streaminfo *mystream, struct mrl_tuple *tuple)
{
    bool ret = false;
    uint8_t type = 0;
    uint16_t  sport = 0, dport = 0, temp_port = 0;
    uint32_t hash_dport =0;
    uint32_t sip = 0, dip = 0;
    sport = tuple->sport;
    dport = tuple->dport;
    type =tuple->protocol;
    sip = ntohl(mystream->addr.tuple4_v4->saddr);
    dip = ntohl(mystream->addr.tuple4_v4->daddr);

    hash_dport = sport ^ sip ^ dip ^type;
    hash_dport = hash_dport & 0xff;
    temp_port = dport & 0xff;
    MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_snat_pkt_identify","cur stream[sip:%s,sport:%hu,dip:%s,dport:%hu,protocol:%hu]"
                            "hash info is:[sip:%u,dip:%u,sport:%hu,type:%hu,hash_sport:%u, ir_port:%u]",
                            tuple->sip,tuple->sport,tuple->dip,tuple->dport,tuple->protocol,
                            sip,dip,sport,type,hash_dport,temp_port);
    if(hash_dport == temp_port)
    {
        ret = true;
    }
    else
    {
        ret = false;
    }
    return ret;
}



long ht_vxlan_info_search_cb(void *data, const uchar *key, uint size, void *user_arg)
{
    struct mrl_vxlan_info *vxlan_info = (struct mrl_vxlan_info *)data;
    uint64_t *link_identity = (uint64_t* )user_arg;
    if(vxlan_info != NULL)
    {
        *link_identity = vxlan_info->link_identity;
        return 1;
    }
    else
    {
        return 0;
    }
}

bool mrl_get_link_identity(const char* back_ip, uint64_t* link_identity)
{
    long ret = 0;
    
    if(MESA_htable_search_cb(mrl_instance.ht_vxlan_info, (const unsigned char *)back_ip, MRL_STR_IP_LEN,ht_vxlan_info_search_cb,link_identity,&ret)!=NULL)
    {
        
         MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_get_link_identity",
                                "cur back_ip %s link identityis:%d",back_ip,*link_identity);
         return true;
    }
    else
    {
        MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_FATAL,"mrl_get_link_identity","can not search the vxlan_info of back_ip %s",back_ip);
        return false;
    }
}

void mrl_parse_link_identity(char* data)
{
    int ret = 0;
    uint64_t identity = 0;
    cJSON* json = NULL;
	cJSON* runtime =NULL;
    cJSON* link_info =NULL;
    cJSON* thread_link_info = NULL;
    cJSON* single_link_info = NULL;
    cJSON* virtual_link_id = NULL;
    json=cJSON_Parse(data);
    if (json == NULL)	
    {
    	return ;
   	}
    runtime=cJSON_GetObjectItemCaseSensitive(json, "runtime");
    if(NULL == runtime)
    {
    	return;
    }
	link_info=cJSON_GetObjectItemCaseSensitive(runtime, "linkinfo");
    cJSON_ArrayForEach(thread_link_info,link_info)
    {
        if(thread_link_info == NULL)
    	{
    		break;
    	}
        cJSON_ArrayForEach(single_link_info,thread_link_info)
        {
            if(single_link_info == NULL)
            {
            	break;
            }
            virtual_link_id = cJSON_GetObjectItemCaseSensitive(single_link_info, "virtual_link_id");
            if(virtual_link_id == NULL)
            {
            	return;
            }
            uint64_t* link_identity = (uint64_t* )calloc(1,sizeof(uint64_t));
			global_stat.malloc_memory += sizeof(uint64_t);
            *link_identity = (uint64_t)virtual_link_id->valuedouble;
            identity = *link_identity;
            ret = mrl_htable_add(mrl_instance.ht_link_identity,(const unsigned char*)link_identity,sizeof(uint64_t),link_identity);
            if(ret == MRL_HTABLE_ADD_SUCCESS)
            {
                global_stat.link_identity_num ++;
                MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_parse_link_identity","ht_link_identity succeed to add link_identity key[link_identity:%lu]!",identity);
            }
            else
            {
                if(ret == MRL_HTABLE_ADD_DUPLICATE)
                {
                    MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_parse_link_identity","the link_identity key[link_identity:%d] is duplicated in ht_link_identity!",identity);
                }
                else
                {
                    MESA_handle_runtime_log(mrl_instance.mrl_log_handle, RLOG_LV_DEBUG,"mrl_parse_link_identity","ht_link_identity fail to add link_identity key[link_identity:%d]!",identity);
                    assert(0);
                }
            }
        }
    }
    if(json != NULL)
    {           
        cJSON_Delete(json);
    }      
}

    
void mrl_read_link_identity(const char* path)
{
    FILE *f;
    long len;
    char *data;
    f=fopen(path,"rb");
    fseek(f,0,SEEK_END);
    len=ftell(f);
    fseek(f,0,SEEK_SET);
    data=(char*)calloc(len+1,sizeof(char));
	global_stat.malloc_memory += (len+1);
	
    fread(data,1,len,f);
    fclose(f);
    mrl_parse_link_identity(data);
	
    free(data);
	global_stat.free_memory +=(len+1);
    
}

void mrl_detect_ip(const char* src_ip, int nominee_type)
{
    int ret = 0;
    struct user_data data;
    strncpy(data.detect_ip,src_ip,MRL_STR_IP_LEN);
    data.nominee_type = nominee_type;
    ret =MESA_htable_iterate(mrl_instance.ht_link_identity,ht_link_identity_iterate_cb,&data);
    assert(ret == 0);
}


void *mrl_detect_action(void *arg)
{
    sleep(10);
    int ret = 0;
    while(1)
    {
        if(mrl_instance.stop_flag == 1)
        {
            break;
        }
        mrl_read_link_identity(mrl_instance.mrl_cfg.link_identity_path);
        
        ret = MESA_htable_iterate(mrl_instance.ht_nominee,ht_nominee_iterate_cb,NULL);
        assert(ret == 0);
        sleep(60);
    }
    return NULL;
}