tango-kni/kni_intercept.c

#include <stdio.h>   
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <net/if.h>
#include <sys/un.h> 
#include <sys/ioctl.h>
#include <net/ethernet.h>
#include <netpacket/packet.h>
#include <linux/if_tun.h>
#include <time.h>
#include <assert.h>

#include "kni_intercept.h"
#include "kni_entry.h"
 



extern "C" int sendpacket_do_checksum(unsigned char* buf,int protocol,int len);
extern "C" int sendpacket_build_ethernet(unsigned char* dst,unsigned char* src,unsigned short type,const unsigned char* payload,int payload_s,unsigned char* buf);
extern "C" unsigned char MESA_dir_reverse(unsigned char route_dir);


size_t add_option(char* buff, size_t size, uint16_t opt_type, uint16_t opt_len, char* opt_cont)
{
	if(size<opt_len+sizeof(uint16_t)*2)
	{
		return 0;
	}
	*((uint16_t*)buff)=opt_type;
	*((uint16_t*)(buff+sizeof(uint16_t))=opt_len;
	memcpy(buff+sizeof(uint16_t)*2, opt_cont, opt_len);
	return opt_len+sizeof(uint16_t)*2;
}
int kni_set_tlvinfo(char* buf, int buflen, struct kni_lqueue_datainfo datainfo)
{
	int tlv_len = 0;

	struct kni_tlv_header *header_info=(struct kni_tlv_header *)buf;
	
	header_info->magic = 0x4d5a;
	header_info->counts = 2;

	tlv_len += sizeof(struct kni_tlv_header);

	tlv_len+=add_option(buf+tlv_len,buflen-tlv_len, KNI_TLV_TYPE_PROTOCOL, (uint16_t)sizeof(int), &(datainfo.protocol));
	tlv_len+=add_option(buf+tlv_len,buflen-tlv_len, KNI_TLV_TYPE_KEYRING_ID, (uint16_t)sizeof(int), &(datainfo.keyring));

	assert(tlv_len=<buflen);

	return tlv_len;

}

int kni_send_fds(int socket,struct kni_repaired_fds to_send_fds)  
{
	int flags=MSG_NOSIGNAL;
    struct msghdr msg = {0};
    struct cmsghdr *cmsg;
    char buf[CMSG_SPACE(KNI_SENDFD_NUM * sizeof(int))]={0}, dup[256]={0};
	
    struct iovec io = { .iov_base = &dup, .iov_len = sizeof(dup) };

	int dup_len = 256;
	int fds[KNI_SENDFD_NUM]={0};
	fds[0] = to_send_fds.client_fd;
	fds[1] = to_send_fds.server_fd;


	dup_len = kni_set_tlvinfo(dup,dup_len,to_send_fds);

    msg.msg_iov = &io;
    msg.msg_iovlen = 1;
    msg.msg_control = buf;
    msg.msg_controllen = sizeof(buf);

    cmsg = CMSG_FIRSTHDR(&msg);
    cmsg->cmsg_level = SOL_SOCKET;
    cmsg->cmsg_type = SCM_RIGHTS;
    cmsg->cmsg_len = CMSG_LEN(KNI_SENDFD_NUM * sizeof(int));

    memcpy ((int *) CMSG_DATA(cmsg), fds, KNI_SENDFD_NUM * sizeof (int));

	if (sendmsg (socket, &msg, flags) < 0)
    {
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"kni_send_fds","sendmsg()error,errno:%d,%s",errno,strerror(errno));
		return -1;
	}

	return 0;
}


int kni_unixdomain_sendinfo()
{
	int ret=0;
	long datainfo_len = 0;
	struct kni_repaired_fds to_send_fds;

	
	struct timespec start, end;
	long elapse=0;

	while(1)
	{
		datainfo_len = sizeof(to_send_fds);
	
		clock_gettime(CLOCK_MONOTONIC, &start);
		
		ret=MESA_lqueue_get_head(g_kni_structinfo.lqueue_send_fds,&to_send_fds,&datainfo_len);
        if(ret==MESA_QUEUE_RET_QEMPTY)
        {
            continue;
        }
        if(ret<0)
        {
			assert(0);
        }

		clock_gettime(CLOCK_MONOTONIC, &end);
							
		elapse=(end.tv_sec-start.tv_sec)*1000000+(end.tv_nsec-start.tv_nsec)/1000;
		FS_operate(g_kni_fs2_info.handler, g_kni_fs2_info.metric_qout_fd, 0, FS_OP_SET, elapse);

		
		clock_gettime(CLOCK_MONOTONIC, &start);

		ret=kni_send_fds(g_kni_comminfo.fd_domain,to_send_fds);
		if(ret<0)	//check errno
		{
			kni_filestate2_set(0,COLUME_TCPREPAIR_SEND_ERR,0,2);
			g_kni_comminfo.kni_mode_cur=KNI_MODE_BYPASS;
			close(to_send_fds.client_fd);
			close(to_send_fds.server_fd);
			return -1;
		}
		kni_filestate2_set(0,COLUME_TCPREPAIR_SEND_SUCC,0,2);
		
		close(to_send_fds.client_fd);
		close(to_send_fds.server_fd);	
		clock_gettime(CLOCK_MONOTONIC, &end);
							
		elapse=(end.tv_sec-start.tv_sec)*1000000+(end.tv_nsec-start.tv_nsec)/1000;
		FS_operate(g_kni_fs2_info.handler, g_kni_fs2_info.metric_sendfd, 0, FS_OP_SET, elapse);

	}

	return 0;
}

int kni_unixdomain_connect()
{
	while(1)
	{
		g_kni_comminfo.fd_domain=kni_unixdomain_create();
		if(g_kni_comminfo.fd_domain<0)
		{
			sleep(1);
		}
		else
		{
			g_kni_comminfo.kni_mode_cur=KNI_MODE_WORK;
			return 0;
		}
	}

	return 0;
}

/********************************************************************************************************************
name:
function:
return:
*********************************************************************************************************************/
void* kni_unixdomain_process(void* arg)
{
	while(1)
	{
		if(g_kni_comminfo.kni_mode_cur==KNI_MODE_BYPASS)
		{
			kni_unixdomain_connect();
		}
		else if(g_kni_switch_info.send_fds_mode == 0)
		{
			kni_unixdomain_sendinfo();
		}
		else
		{
			sleep(10);
		}
	}

	return 0;

}




int kni_unixdomain_create()
{
	
	int i_fd = 0;    
	struct sockaddr_un addr;    
	int i_addr_len = sizeof( struct sockaddr_un );    

	if ( ( i_fd = socket( AF_UNIX, SOCK_STREAM, 0 ) ) < 0 )  		
//	if ( ( i_fd = socket( AF_UNIX, SOCK_DGRAM, 0 ) ) < 0 )  
	{    
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_INIT,"kni_unixdomain_create():socket error,errno is %d,%s,,action:%s",errno,strerror(errno),KNI_ACTION_EXIT);
		return -1;
	}

//fill socket adress structure with server's address    
	memset( &addr, 0, sizeof( addr ) );    
	addr.sun_family = AF_UNIX;
//	strncpy( addr.sun_path, serverpath, sizeof( addr.sun_path ) - 1 );    
	strncpy( addr.sun_path, g_kni_comminfo.domain_path, sizeof( addr.sun_path ) - 1 );	  


	if ( connect( i_fd, ( struct sockaddr * )&addr, i_addr_len ) < 0 )  
	{
		close(i_fd);
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_INIT,"kni_unixdomain_create():connect error,errno is %d,%s,action:%s",errno,strerror(errno),KNI_ACTION_EXIT);
		return -1;
	}

	
	MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"kni_unixdomain_create","domain socket connect succ! ");
	return i_fd;
}



int init_kni_unixdomain()
{
	pthread_t pid_pro_domain;

	g_kni_comminfo.fd_domain=kni_unixdomain_create();
	if(g_kni_comminfo.fd_domain<0)
	{
		g_kni_comminfo.kni_mode_cur=KNI_MODE_BYPASS;
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_INIT,"kni_unixdomain_create()error");
	}

	pthread_create(&pid_pro_domain,NULL,kni_unixdomain_process,NULL);

	return 0;

}


int tun_set_queue(int fd, int enable)
 {
	 struct ifreq ifr;

	 memset(&ifr, 0, sizeof(ifr));

	 if (enable)
		ifr.ifr_flags = IFF_ATTACH_QUEUE;
	 else
		ifr.ifr_flags = IFF_DETACH_QUEUE;

	 return ioctl(fd, TUNSETQUEUE, (void *)&ifr);
 }

int tun_error(int i,int* fds)
{
	
	for (--i; i >= 0; i--)
	{
		close(fds[i]);
	}
	
	return 0;
}



/* Flags: IFF_TUN	- TUN device (no Ethernet headers)
 *		  IFF_TAP	- TAP device
 *
 *		  IFF_NO_PI - Do not provide packet information
 *		  IFF_MULTI_QUEUE - Create a queue of multiqueue device
 */
int tun_alloc_mq(char *dev, int queues, int *fds,char* tun_path)
{
	int i=0;
	int err=0;
	int fd;
	int flag=0;
	struct ifreq ifr;
	

//	char *clonedev = (char*)"/dev/net/tun";

	memset(&ifr, 0, sizeof(ifr));
	ifr.ifr_flags = IFF_TUN | IFF_NO_PI | IFF_MULTI_QUEUE;		
	 if (*dev) 
	 {
     	strncpy(ifr.ifr_name, dev, IFNAMSIZ);
   }

	for (i = 0; i < queues; i++) 
	{
		if ((fd = open(tun_path, O_RDWR)) < 0)
		{
			MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_INIT,"tun_alloc_mq():open error,errno is:%d,%s,action:%s",errno,strerror(errno),KNI_ACTION_EXIT);
			tun_error(i,fds);
			return -1;
		}
		err = ioctl(fd, TUNSETIFF, (void *)&ifr);
		if (err) 
		{
			MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_INIT,"tun_alloc_mq():ioctl error,errno is:%d,%s,action:%s",errno,strerror(errno),KNI_ACTION_EXIT);
			close(fd);
			tun_error(i,fds);
			return -1;
		}

//20180618 add set noblock
		flag= fcntl(fd, F_GETFL, 0);
		if(flag<0)
		{
			MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_INIT,"fcntl():getfl error,errno is:%d,%s",errno,strerror(errno));
		}

		if( fcntl( fd, F_SETFL, flag|O_NONBLOCK ) < 0 )
		{
			MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_INIT,"fcntl():setfl error,errno is:%d,%s",errno,strerror(errno));
		}

//end

		fds[i] = fd;
	}
	return 0;
}



 int init_kni_tun()
 {
	int ret;
	
	char tun_path[KNI_CONF_MAXLEN]={0};
	
	MESA_load_profile_string_def((char*)KNI_CONF_FILENAME,(char*)KNI_TUN_MODE,(char*)"tun_path",tun_path,KNI_CONF_MAXLEN,"/dev/net/tun");	
	MESA_load_profile_string_def((char*)KNI_CONF_FILENAME,(char*)KNI_TUN_MODE,(char*)"tun_name",g_kni_comminfo.tun_name,KNI_CONF_MAXLEN,"tun0");

	
//	g_kni_comminfo.tun_threadnum=g_iThreadNum;
	if(g_kni_comminfo.tun_threadnum<=0)
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_INIT,"thread_num:%d,action:%s",g_kni_comminfo.thread_num,KNI_ACTION_EXIT);
		return -1;
	}

	g_kni_comminfo.fd_tun=ALLOC(int, g_kni_comminfo.tun_threadnum);
	
	ret=tun_alloc_mq(g_kni_comminfo.tun_name,g_kni_comminfo.tun_threadnum,g_kni_comminfo.fd_tun,tun_path);

	return ret;
		
}


char kni_add_lqueue(int addrtype,int thread_seq,char* send_buf,int send_buflen,const struct streaminfo* pstream,int index)
{
	
 	char ret=APP_STATE_DROPPKT|APP_STATE_GIVEME;	
	int listq_ret = 0;	
	struct kni_inject_pkt to_inject;

	to_inject.addr_type = addrtype;
	to_inject.buflen = send_buflen;
	to_inject.buf = ALLOC(char, send_buflen);
	memcpy(to_inject.buf,send_buf,send_buflen);
	clock_gettime(CLOCK_MONOTONIC, &(to_inject.start));
	
	listq_ret=MESA_lqueue_join_tail(g_kni_structinfo.lqueue_write_tun[index],(void*)&to_inject,sizeof(to_inject));
	if(listq_ret <0)
	{
		kni_htable_del(pstream,(const void*)send_buf);

		free(to_inject.buf);
		to_inject.buf = NULL;

		ret=APP_STATE_DROPPKT|APP_STATE_DROPME;
		kni_filestate2_set(thread_seq,COLUME_ADD_LQUEUE_ERR,0,1);
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_SENDFD,"kni_add_lqueue() error,ret:%d",ret);
		return ret;
	}

	kni_filestate2_set(thread_seq,COLUME_ADD_LQUEUE_SUCC,0,1);

	return ret;
	
}


int kni_sendpkt_sockraw(int thread_seq,int iplen,char* ip,struct stream_tuple4_v4* ipv4_addr,int iprever_flag,int routdir,uchar* smac,uchar* dmac)
{
	if((routdir!=0)&&(routdir!=1))
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_DEBUG,"kni_sendpkt_sockraw","routdir:%d",routdir);
		return -1;
	}

	int ret=0;
	int buflen=iplen+KNI_ETHER_LEN;

	unsigned char buf[2000]={0};
	unsigned short eth_type=0x0800;

	uchar* tmp_smac;
	uchar* tmp_dmac;

	
	struct kni_ipv6_hdr* ipv6_hdr = (struct kni_ipv6_hdr*)buf;
	unsigned short eth_type_v6 = 0x86dd;

	char* if_name=NULL;
	if(iprever_flag==0)
	{
		tmp_smac=smac;
		tmp_dmac=dmac;
	}
	else
	{
		tmp_smac=dmac;
		tmp_dmac=smac;
	}

	if_name=g_kni_cardname[1-routdir];

	struct ifreq ifr;
	size_t ifname_len=strlen(if_name);
	if(ifname_len<sizeof(ifr.ifr_name))
	{
		memset(ifr.ifr_name,0,IFNAMSIZ);
		assert(ifname_len<=IFNAMSIZ);
		memcpy(ifr.ifr_name,if_name,ifname_len);
	}
	else
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger,RLOG_LV_FATAL,KNI_MODULE_SENDPKT,"interface name :%s is too long\n",if_name);
		return -1;
	}

	if(-1==ioctl(g_kni_comminfo.fd_sendpkt[thread_seq],SIOCGIFINDEX,&ifr))
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger,RLOG_LV_FATAL,KNI_MODULE_SENDPKT,"get if index error:%d,%s,name:%d",errno,strerror(errno),if_name);
		return -1;
	}

	struct sockaddr_ll addr={0};
	addr.sll_family=AF_PACKET;
	addr.sll_halen=ETHER_ADDR_LEN;
	addr.sll_ifindex=ifr.ifr_ifindex;
	addr.sll_protocol=htons(ETH_P_IP);
	assert(sizeof(addr.sll_addr)>=ETHER_ADDR_LEN);
	memcpy(addr.sll_addr,tmp_dmac,ETHER_ADDR_LEN);


	if(ioctl(g_kni_comminfo.fd_sendpkt[thread_seq],SIOCGIFHWADDR,&ifr)==-1)
	{
		return -1;
	}

	if((ipv6_hdr->ip6_flags[0] & 0xF0) == 0x60)
	{
		sendpacket_build_ethernet((unsigned char*)tmp_dmac,(unsigned char*)tmp_smac,eth_type_v6,(const unsigned char*)ip,iplen,(unsigned char*)buf);	

	}
	else
	{
		sendpacket_build_ethernet((unsigned char*)tmp_dmac,(unsigned char*)tmp_smac,eth_type,(const unsigned char*)ip,iplen,(unsigned char*)buf);	
	}


	ret=sendto(g_kni_comminfo.fd_sendpkt[thread_seq],buf,buflen,0,(struct sockaddr*)&addr,sizeof(addr));
	if(ret<0)
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger,RLOG_LV_FATAL,KNI_MODULE_SENDPKT,"sendto() error,errno:%d,msg:%s!",errno,strerror(errno));
	}

	return ret;

}





int kni_keepalive_replay_v6(struct stream_tuple4_v6* ipv6_addr,int iprever_flag,struct kni_htable_datainfo* datainfo,void* a_packet,int iplen,int thread_seq)
{
	if(!g_kni_switch_info.replay_win_update)
	{
		return 0;
	}

	int index=1-iprever_flag;

	struct kni_ipv6_hdr* ipv6_hdr = (struct kni_ipv6_hdr*)a_packet;

//ipv6 has opt not process  20181115 modify	
	if((ipv6_hdr != NULL) && (ipv6_hdr->ip6_nex_hdr != NEXTHDR_TCP))
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger,RLOG_LV_FATAL,KNI_MODULE_SENDPKT,"kni_keepalive_replay_v6():ipv6_hdr->ip6_nex_hdr != NEXTHDR_TCP,not send window probe replay");
		return 0;
	}
//end
	
	struct kni_tcp_hdr* tcphdr=(struct kni_tcp_hdr*)((unsigned char*)a_packet + sizeof(struct kni_ipv6_hdr));
	
	struct kni_wndpro_reply_info* tcpinfo=&(datainfo->lastpkt_info[index]);

	struct kni_ipv6_hdr* snd_iphdr=NULL;
	struct kni_tcp_hdr* snd_tcphdr=NULL;
	char* sendbuf= ALLOC(char, iplen);
	memcpy(sendbuf, a_packet, iplen);

	snd_iphdr=(struct kni_ipv6_hdr*)sendbuf;
	snd_tcphdr=(struct kni_tcp_hdr*)((unsigned char*)snd_iphdr + sizeof(struct kni_ipv6_hdr));

	snd_iphdr->ip6_src=ipv6_hdr->ip6_dst;
	snd_iphdr->ip6_dst=ipv6_hdr->ip6_src;

	snd_tcphdr->th_sport=tcphdr->th_dport;
	snd_tcphdr->th_dport=tcphdr->th_sport;
	snd_tcphdr->th_seq=htonl(tcpinfo->seq+tcpinfo->len);
	snd_tcphdr->th_ack=htonl(tcpinfo->ack);

	
	if(tcpinfo->syn_flag==1)
	{
		snd_tcphdr->th_seq=htonl(ntohl(snd_tcphdr->th_seq)+1);
	}
	sendpacket_do_checksum((unsigned char*)sendbuf,IPPROTO_TCP,htons(ipv6_hdr->ip6_payload_len));
//	sendpacket_do_checksum((unsigned char*)sendbuf,IPPROTO_IP,sizeof(struct kni_ipv6_hdr));

//	tun_write_data(g_kni_comminfo.fd_tun[thread_seq],sendbuf,iplen,NULL,thread_seq);
	tun_write_data(g_kni_comminfo.fd_tun[thread_seq],sendbuf,iplen,NULL,thread_seq);


	
	kni_log_debug(RLOG_LV_DEBUG,(char*)"win_update",a_packet,(char*)"recv tcp_repair windows update,and replay");

	free(sendbuf);
	sendbuf=NULL;

	datainfo->wndprob_flag[index]=1;

	kni_filestate2_set(thread_seq,COLUME_REPLAY_WINDOW,0,1);
	

	return 1;
	
	
}



int kni_keepalive_replay(struct stream_tuple4_v4* ipv4_addr,int iprever_flag,struct kni_htable_datainfo* datainfo,void* a_packet,int iplen,int thread_seq)
{

	if(!g_kni_switch_info.replay_win_update)
	{
		return 0;
	}


	int index=1-iprever_flag;
	
	struct ip* iphdr=(struct ip*)a_packet;
	struct tcphdr* tcphdr=(struct tcphdr*)((char*)iphdr+4*(iphdr->ip_hl));
	struct kni_wndpro_reply_info* tcpinfo=&(datainfo->lastpkt_info[index]);

	struct ip* snd_iphdr=NULL;
	struct tcphdr* snd_tcphdr=NULL;
	char* sendbuf=ALLOC(char, iplen);
	memcpy(sendbuf,a_packet,iplen);

	snd_iphdr=(struct ip*)sendbuf;
	snd_tcphdr=(struct tcphdr*)((char*)snd_iphdr+4*(snd_iphdr->ip_hl));

	(snd_iphdr->ip_src).s_addr=(iphdr->ip_dst).s_addr;
	(snd_iphdr->ip_dst).s_addr=(iphdr->ip_src).s_addr;

	snd_tcphdr->source=tcphdr->dest;
	snd_tcphdr->dest=tcphdr->source;
	snd_tcphdr->seq=htonl(tcpinfo->seq+tcpinfo->len);
	snd_tcphdr->ack_seq=htonl(tcpinfo->ack);

	
	if(tcpinfo->syn_flag==1)
	{
		snd_tcphdr->seq=htonl(ntohl(snd_tcphdr->seq)+1);
	}

	sendpacket_do_checksum((unsigned char*)sendbuf,IPPROTO_TCP,(iplen-4*(iphdr->ip_hl)));
	sendpacket_do_checksum((unsigned char*)sendbuf,IPPROTO_IP,sizeof(struct ip));

//	tun_write_data(g_kni_comminfo.fd_tun[thread_seq],sendbuf,iplen,NULL,thread_seq);	
	tun_write_data(g_kni_comminfo.fd_tun[thread_seq],sendbuf,iplen,NULL,thread_seq);
	
	kni_log_debug(RLOG_LV_DEBUG,(char*)"win_update",a_packet,(char*)"recv tcp_repair windows update,and replay");

	free(sendbuf);
	sendbuf=NULL;

	datainfo->wndprob_flag[index]=1;

	kni_filestate2_set(thread_seq,COLUME_REPLAY_WINDOW,0,1);
	

	return 1;
	
	
}



long kni_readtun_htable_cb_v6(void* data,const unsigned char* key,unsigned int size,void* user_arg)
{
	long result=0;
	struct stream_tuple4_v6* ipv6_addr=(struct stream_tuple4_v6*)key;
	struct args_read_tun* args=(struct args_read_tun*)user_arg;
	struct kni_htable_datainfo* datainfo=(struct kni_htable_datainfo*)data;
	
	if(datainfo!=NULL) 
	{
		memcpy(args->smac, datainfo->smac, sizeof(args->smac));
		memcpy(args->dmac, datainfo->dmac, sizeof(args->dmac));
		if(args->iprevers==0)
		{
			args->routdir=datainfo->route_dir;
		}
		else
		{
			if(g_kni_switch_info.sendpkt_mode == 1)
			{
				args->routdir=1-datainfo->route_dir;

			}
			else
			{
				args->routdir=MESA_dir_reverse(datainfo->route_dir);
			}
		}

		if(datainfo->wndprob_flag[1-args->iprevers]>0)
		{
			result=1;
		}
		else
		{
			kni_keepalive_replay_v6(ipv6_addr,args->iprevers,datainfo,args->a_packet,args->iplen,args->thread_seq);
			result=1;
		}
	}
	
	return result;
}



long kni_readtun_htable_cb_v4(void* data,const unsigned char* key,unsigned int size,void* user_arg)
{
	long result=0;
	struct stream_tuple4_v4* ipv4_addr=(struct stream_tuple4_v4*)key;
	struct args_read_tun* args=(struct args_read_tun*)user_arg;
	struct kni_htable_datainfo* datainfo=(struct kni_htable_datainfo*)data;
	
	if(datainfo!=NULL) 
	{
		memcpy(args->smac, datainfo->smac, sizeof(args->smac));
		memcpy(args->dmac, datainfo->dmac, sizeof(args->dmac));
		if(args->iprevers==0)
		{
			args->routdir=datainfo->route_dir;
		}
		else
		{
			if(g_kni_switch_info.sendpkt_mode == 1)
			{
				args->routdir=1-datainfo->route_dir;

			}
			else
			{
				args->routdir=MESA_dir_reverse(datainfo->route_dir);
			}

		}

		if(datainfo->wndprob_flag[1-args->iprevers]>0)
		{
			result=1;
		}
		else
		{
			kni_keepalive_replay(ipv4_addr,args->iprevers,datainfo,args->a_packet,args->iplen,args->thread_seq);
			result=1;
		}
		
		
		
	}
#ifdef KNI_DEBUG_TCPREPAIR
	else if(ipv4_addr->saddr==1698867392)
	{
		printf("sip is 192.168.66.101\n");
		args->smac[0]=0x18;
		args->smac[1]=0x66;
		args->smac[2]=0xda;
		args->smac[3]=0xe5;
		args->smac[4]=0xfa;
		args->smac[5]=0xa1;

		args->dmac[0]=0xe8;
		args->dmac[1]=0x61;
		args->dmac[2]=0x1f;
		args->dmac[3]=0x13;
		args->dmac[4]=0x70;
		args->dmac[5]=0x7a;
		result=1;
	}
	else
	{
		kni_log_debug(RLOG_LV_INFO,"htable_cb",(void*)args->a_packet,(char*)"read_tun,data=NULL,sip not 192.168.66.101");
		
	}
#endif
	return result;
}





int kni_process_readdata(int thread_seq,int sendpkt_threadid,int buflen,char* buf)
{
	int iprever_flag=0;
	long result=0;

	struct ip* ipv4_hdr=(struct ip*)buf;
	struct kni_ipv6_hdr* ipv6_hdr = (struct kni_ipv6_hdr*)buf;
	struct stream_tuple4_v4 ipv4_addr;
	struct stream_tuple4_v6 ipv6_addr;

	struct args_read_tun args;
	
	if(ipv4_hdr->ip_v==4)
	{
		iprever_flag=kni_get_ipaddr_v4((void*)buf,&ipv4_addr);

		args.a_packet=buf;
		args.iplen=buflen;
		args.iprevers=iprever_flag;
		args.thread_seq=thread_seq;

		MESA_htable_search_cb(g_kni_structinfo.htable_to_tun_v4,(unsigned char*)&ipv4_addr,sizeof(struct stream_tuple4_v4),kni_readtun_htable_cb_v4,(void*)&args,&result);
		if(result==1)
		{
			kni_filestate2_set(thread_seq,COLUME_SEND_PKT,0,1);
		
			if(g_kni_switch_info.sendpkt_mode == 1)
			{
				kni_sendpkt_sockraw(thread_seq,buflen,buf,&ipv4_addr,iprever_flag,args.routdir,args.smac,args.dmac);
			}
			else
			{
				MESA_sendpacket_iplayer_options(thread_seq,buf,buflen,args.routdir,NULL,0);
			}
		}
		else
		{
			kni_log_debug(RLOG_LV_FATAL,(char*)"kni_readtun_htable_cb_v4",buf,(const char*)"kni_readtun_htable_cb_v4 not found!");
			kni_filestate2_set(thread_seq,COLUME_DROP_NOIN_HTABLE,0,1);
		}

	}
	else if((ipv6_hdr->ip6_flags[0] & 0xF0) == 0x60)
	{
		iprever_flag=kni_get_ipaddr_v6((void*)buf,&ipv6_addr);

		args.a_packet=buf;
		args.iplen=buflen;
		args.iprevers=iprever_flag;
		args.thread_seq=thread_seq;

		MESA_htable_search_cb(g_kni_structinfo.htable_to_tun_v6,(unsigned char*)&ipv6_addr,sizeof(struct stream_tuple4_v6),kni_readtun_htable_cb_v6,(void*)&args,&result);
		if(result==1)
		{
			kni_filestate2_set(thread_seq,COLUME_SEND_PKT,0,1);
			
			if(g_kni_switch_info.sendpkt_mode == 1)
			{
				kni_sendpkt_sockraw(thread_seq,buflen,buf,&ipv4_addr,iprever_flag,args.routdir,args.smac,args.dmac);
			}
			else
			{
				MESA_sendpacket_ipv6_layer_options(thread_seq,buf,buflen,args.routdir,NULL,0);
			}

		}
		else
		{
			kni_filestate2_set(thread_seq,COLUME_DROP_NOIN_HTABLE,0,1);
		}
		
	}
	else
	{
		kni_filestate2_set(thread_seq,COLUME_DROP_NOTIPV46_TUN,0,1);
		MESA_handle_runtime_log(g_kni_comminfo.logger,RLOG_LV_FATAL,KNI_MODULE_SENDPKT,"kni_readdata,not ipv4 and not ipv6!");
			
	}

	return 0;

}

int kni_process_writedata(int thread_seq)
{
	int ret = 0;	
	struct kni_inject_pkt datainfo;
	long datainfo_len = sizeof(datainfo);

	struct timespec end;
	long elapse=0;
	
	ret=MESA_lqueue_get_head(g_kni_structinfo.lqueue_write_tun[thread_seq],&datainfo,&datainfo_len);
	if(ret == MESA_QUEUE_RET_QEMPTY)
	{
		return 0;
	}
	else if(ret<0)
	{
		kni_filestate2_set(thread_seq,COLUME_GET_LQUEUE_ERR,0,1);
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,(char*)"kni_process_writedata","MESA_lqueue_try_get_tail() error!ret:%d,datalen:%d\n",ret,datainfo_len);
		return -1;
	}
	else
	{
		
		clock_gettime(CLOCK_MONOTONIC, &end);
									
		elapse=(end.tv_sec-datainfo.start.tv_sec)*1000000+(end.tv_nsec-datainfo.start.tv_nsec)/1000;
		FS_operate(g_kni_fs2_info.handler, g_kni_fs2_info.metric_qout_pkt,0,FS_OP_SET, elapse);

	
		kni_filestate2_set(thread_seq,COLUME_GET_LQUEUE_SUCC,0,1);
		
		tun_write_data_listq(g_kni_comminfo.fd_tun[thread_seq],datainfo.buf,datainfo.buflen,thread_seq);

		free(datainfo.buf);
		datainfo.buf = NULL;
	}

	return 0;
}


int tun_read_data(int fd, char* buffer, size_t size)
{
	int recv_len=0;

	int ret=0;
    int max_fd = 0;

	if(g_kni_switch_info.write_listq_switch == 0)
	{
		fd_set alive_readfd;
		struct timeval timeout;
		memset(&timeout,0,sizeof(timeout));
	
		
		FD_ZERO(&alive_readfd);
		FD_SET(fd, &alive_readfd);
		max_fd = fd;
		
	//	ret = select(max_fd + 1, &alive_readfd, NULL, NULL, &timeout);
		ret = select(max_fd + 1, &alive_readfd, NULL, NULL, NULL);
		if (ret < 0)
		{
	//		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL, "keep_alive_action function", "select function errno %d is %s!", errno, strerror(errno));
			return 0;
		}
	}
	
	recv_len = read(fd, buffer, size);
	if(recv_len <0)
	{
//		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_WRITETUN,"tun_read_data error %d, %s\n",errno,strerror(errno));
		return -1;
	}
	else
	{
		return recv_len;
	}

	return 0;
}



char tun_write_data_listq(int fd,char* send_buf,int send_buflen,int thread_seq)
{
 	char ret=APP_STATE_DROPPKT|APP_STATE_GIVEME;	
	int succ_sendlen=0;
	
	struct timespec start, end;
	long elapse=0;
	clock_gettime(CLOCK_MONOTONIC, &start);
	
	succ_sendlen = write(fd, send_buf,send_buflen);
	if(succ_sendlen<0)
	{
		kni_filestate2_set(thread_seq,COLUME_WRITE_TUN_ERR,0,1);
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_WRITETUN,"write() error %d, %s",errno,strerror(errno));
		ret=APP_STATE_DROPPKT|APP_STATE_DROPME;
	}
	else if(succ_sendlen<send_buflen)
	{
		kni_filestate2_set(thread_seq,COLUME_WRITE_TUN_ERR,0,1);
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_WRITETUN,"succ_sendlen is %d,send_buflen is %d",succ_sendlen,send_buflen);
	}
	else
	{
		kni_filestate2_set(thread_seq,COLUME_WRITE_TUN_SUCC,0,1);
	}

	kni_filestate2_set(thread_seq,COLUME_TUN_WRITE,0,1);

	clock_gettime(CLOCK_MONOTONIC, &end);
					
	elapse=(end.tv_sec-start.tv_sec)*1000000+(end.tv_nsec-start.tv_nsec)/1000;
	FS_operate(g_kni_fs2_info.handler, g_kni_fs2_info.metric_tun_write, 0, FS_OP_SET, elapse);

	return ret;
	
}




char tun_write_data(int fd, const char* send_buf, size_t send_buflen,struct streaminfo* pstream,int thread_seq)
{
 	char ret=APP_STATE_DROPPKT|APP_STATE_GIVEME;	
	int succ_sendlen=0;
	
	struct timespec start, end;
	long elapse=0;
	clock_gettime(CLOCK_MONOTONIC, &start);
	
	succ_sendlen = write(fd, send_buf,send_buflen);
	if((succ_sendlen<0)&&(pstream!=NULL))
	{
		kni_htable_del(pstream,(const void*)send_buf);
	
		kni_filestate2_set(thread_seq,COLUME_WRITE_TUN_ERR,0,1);
//		MESA_kill_tcp(pstream,(const void*)send_buf);
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_WRITETUN,"write() error %d, %s",errno,strerror(errno));
		ret=APP_STATE_DROPPKT|APP_STATE_DROPME;
	}
	else if(succ_sendlen<send_buflen)
	{
		kni_htable_del(pstream,(const void*)send_buf);
	
		kni_filestate2_set(thread_seq,COLUME_WRITE_TUN_ERR,0,1);
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_WRITETUN,"succ_sendlen is %d,send_buflen is %d",succ_sendlen,send_buflen);
	}

	kni_filestate2_set(thread_seq,COLUME_TUN_WRITE,0,1);

	clock_gettime(CLOCK_MONOTONIC, &end);
					
	elapse=(end.tv_sec-start.tv_sec)*1000000+(end.tv_nsec-start.tv_nsec)/1000;
	FS_operate(g_kni_fs2_info.handler, g_kni_fs2_info.metric_tun_write, 0, FS_OP_SET, elapse);

	return ret;
	
}



int tun_write_data_v6(int fd,char* send_buf,int send_buflen)
{
	int succ_sendlen=0;
	
	succ_sendlen = write(fd, send_buf,send_buflen);
	if(succ_sendlen<0)
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_WRITETUN,"write() error!msg is %s",strerror(errno));
	}
	else if(succ_sendlen<send_buflen)
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_WRITETUN,"succ_sendlen is %d,send_buflen is %d",succ_sendlen,send_buflen);
	}

	MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_DEBUG,KNI_MODULE_WRITETUN,"write to tun completed,,send_buflen is %d",succ_sendlen);

	return succ_sendlen;
	
}



/********************************************************************************************************************
name:
function:
return:
*********************************************************************************************************************/
void* pthread_process_tun(void* arg)
{
	int thread_seq=*(int*)arg;
	int recv_len=0;
	char recv_buf[KNI_MAX_BUFLEN] = {0};

	int sendpkt_threadid=0;
	int sendpkt_threadid_len = sizeof(int);
	long elapse=0;
	sapp_get_platform_opt(SPO_INDEPENDENT_THREAD_ID,&sendpkt_threadid,&sendpkt_threadid_len);

	struct timespec start, end;

	while(1)
	{
		if(g_kni_comminfo.kni_mode_cur==KNI_MODE_BYPASS)
		{
			sleep(KNI_USLEEP_TIME);
			continue;
		}
//write to run
		if(g_kni_switch_info.write_listq_switch == 1)
		{
			kni_process_writedata(thread_seq);
		}

//end
		
//read from tun
		clock_gettime(CLOCK_MONOTONIC, &start);
		recv_len=0;
		recv_len=tun_read_data(g_kni_comminfo.fd_tun[thread_seq], recv_buf, sizeof(recv_buf));
		if(recv_len <0)
		{
//			MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_READTUN,"tun_read_data()error,recv_len:%d",recv_len);
		}
		else if(recv_len>0)
		{
		
			clock_gettime(CLOCK_MONOTONIC, &end);
			
			elapse=(end.tv_sec-start.tv_sec)*1000000+(end.tv_nsec-start.tv_nsec)/1000;
			FS_operate(g_kni_fs2_info.handler, g_kni_fs2_info.metric_tun_read, 0, FS_OP_SET, elapse);
			start=end;
			kni_filestate2_set(thread_seq,COLUME_TUN_READ,0,1);
			kni_process_readdata(thread_seq,sendpkt_threadid,recv_len,recv_buf);
			clock_gettime(CLOCK_MONOTONIC, &end);
			elapse=(end.tv_sec-start.tv_sec)*1000000+(end.tv_nsec-start.tv_nsec)/1000;
			
			FS_operate(g_kni_fs2_info.handler, g_kni_fs2_info.metric_forward, 0, FS_OP_SET, elapse);
		}
//end
	}

	return 0;

}




int tcprepair_get_addr(void** client_addr,void** server_addr,const struct layer_addr* addr,const void* a_packet)
{
	struct ip* ipv4_hdr = NULL;
	struct kni_ipv6_hdr* ipv6_hdr = NULL;
	struct kni_tcp_hdr* tcphdr=NULL;

	struct sockaddr_in* client_addr_v4;
	struct sockaddr_in* server_addr_v4;

	struct sockaddr_in6* client_addr_v6;
	struct sockaddr_in6* server_addr_v6;

	if(addr->addrtype == ADDR_TYPE_IPV4)
	{
		ipv4_hdr = (struct ip*)a_packet;
		tcphdr=(struct kni_tcp_hdr*)((char*)ipv4_hdr+4*(ipv4_hdr->ip_hl));
	
		client_addr_v4 = ALLOC(struct sockaddr_in, 1);
		client_addr_v4->sin_family = AF_INET;
		client_addr_v4->sin_port = tcphdr->th_sport;
		client_addr_v4->sin_addr.s_addr = (ipv4_hdr->ip_src).s_addr;
		
		server_addr_v4 = ALLOC(struct sockaddr_in, 1);
		server_addr_v4->sin_family = AF_INET;
		server_addr_v4->sin_port = tcphdr->th_dport;
		server_addr_v4->sin_addr.s_addr = (ipv4_hdr->ip_dst).s_addr;

		*client_addr = client_addr_v4;
		*server_addr = server_addr_v4;
	}
	else if(addr->addrtype == ADDR_TYPE_IPV6)
	{
		ipv6_hdr = (struct kni_ipv6_hdr*)a_packet;
		tcphdr = (struct kni_tcp_hdr*)((unsigned char*)a_packet + sizeof(struct kni_ipv6_hdr));
	
		client_addr_v6 = ALLOC(struct sockaddr_in6, 1);
		client_addr_v6->sin6_family = AF_INET6;
		client_addr_v6->sin6_port = tcphdr->th_sport;
		
		server_addr_v6 = ALLOC(struct sockaddr_in6, 1);
		server_addr_v6->sin6_family = AF_INET6;
		server_addr_v6->sin6_port = tcphdr->th_dport;
		server_addr_v6->sin6_addr=ipv6_hdr->ip6_dst;

		*client_addr = client_addr_v6;
		*server_addr = server_addr_v6;
	}
	else
	{
		return -1;
	}
	
	return 0;
}



int tcprepair_free_addr(struct sockaddr* client_addr,struct sockaddr* server_addr)
{
	if(client_addr != NULL)
	{
		free(client_addr);
		client_addr =NULL;
	}

	if(server_addr != NULL)
	{
		free(server_addr);
		server_addr = NULL;
	}

	return 0;
}


int tcprepair_get_state(int curdir,const struct layer_addr* addr,struct kni_tcp_state* fake_client,struct kni_tcp_state* fake_server,const void* a_packet,struct kni_pme_info* pmeinfo)
{

	struct ip* ipv4_hdr = NULL;
	struct kni_tcp_hdr* tcphdr=NULL;

	if(addr->addrtype==ADDR_TYPE_IPV4)
	{
		ipv4_hdr = (struct ip*)a_packet;
		tcphdr=(struct kni_tcp_hdr*)((char*)ipv4_hdr+4*(ipv4_hdr->ip_hl));
	}
	else if(addr->addrtype==ADDR_TYPE_IPV6)
	{
		tcphdr = (struct kni_tcp_hdr*)((unsigned char*)a_packet + sizeof(struct kni_ipv6_hdr));
	}
	else
	{
		assert(0);
	}
	
	fake_client->seq=ntohl(tcphdr->th_seq);
	fake_client->ack=ntohl(tcphdr->th_ack);
	fake_client->mss_src=pmeinfo->tcpopt_info[KNI_DIR_C2S].mss;
	fake_client->mss_dst=pmeinfo->tcpopt_info[KNI_DIR_S2C].mss;
	fake_client->wscale_src=pmeinfo->tcpopt_info[KNI_DIR_C2S].wnscal;
	fake_client->wscale_dst=pmeinfo->tcpopt_info[KNI_DIR_S2C].wnscal;
	fake_client->sack_src=pmeinfo->tcpopt_info[KNI_DIR_C2S].sack;
	fake_client->sack_dst=pmeinfo->tcpopt_info[KNI_DIR_S2C].sack;
	fake_client->timestamps_src=pmeinfo->tcpopt_info[KNI_DIR_C2S].timestamps;
	fake_client->timestamps_dst=pmeinfo->tcpopt_info[KNI_DIR_S2C].timestamps;

	fake_server->seq=ntohl(tcphdr->th_ack);
	fake_server->ack=ntohl(tcphdr->th_seq);
	fake_server->mss_src=pmeinfo->tcpopt_info[KNI_DIR_S2C].mss;
	fake_server->mss_dst=pmeinfo->tcpopt_info[KNI_DIR_C2S].mss;
	fake_server->wscale_src=pmeinfo->tcpopt_info[KNI_DIR_S2C].wnscal;
	fake_server->wscale_dst=pmeinfo->tcpopt_info[KNI_DIR_C2S].wnscal;
	fake_server->sack_src=pmeinfo->tcpopt_info[KNI_DIR_S2C].sack;
	fake_server->sack_dst=pmeinfo->tcpopt_info[KNI_DIR_C2S].sack;
	fake_server->timestamps_src=pmeinfo->tcpopt_info[KNI_DIR_S2C].timestamps;
	fake_server->timestamps_dst=pmeinfo->tcpopt_info[KNI_DIR_S2C].timestamps;

	return 0;
}



int tcprepair_set_state(int sk,struct kni_tcp_state* tcp,struct sockaddr* client_addr,struct sockaddr* server_addr,int addr_type)
{
	int val,yes=1, onr = 0;
	int temp_mark=0;

	socklen_t mark_len =sizeof(temp_mark);

	struct tcp_repair_opt opts[KNI_TCPREPAIR_OPT_NUM];
	
	if (setsockopt(sk, SOL_TCP, TCP_REPAIR, &yes, sizeof(yes))==-1)
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","setsockopt() TCP_REPAIR error,errno:%d,%s",errno,strerror(errno));
		return -1;
	}

	
    if (setsockopt(sk, SOL_IP, IP_TRANSPARENT, &yes, sizeof(yes)) < 0) 
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","setsockopt() IP_TRANSPARENT error,errno:%d,%s",errno,strerror(errno));
		return -1;
	}

	
	if (setsockopt(sk, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) == -1)
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","setsockopt() SO_REUSEADDR error,errno:%d,%s",errno,strerror(errno));
		return -1;
	}
	if(setsockopt(sk,SOL_SOCKET,SO_MARK,&(g_kni_comminfo.mark),sizeof(g_kni_comminfo.mark))==-1)
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","setsockopt() SO_MARK error,errno:%d,%s",errno,strerror(errno));
		return -1;

	}
	getsockopt(sk,SOL_SOCKET,SO_MARK,&temp_mark,&mark_len);
	MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_DEBUG,"tcprepair_set_state","setsockopt() fd :%d,SO_MARK:%d",sk,temp_mark);
	/* ============= Restore TCP properties ==================*/
	val = TCP_SEND_QUEUE;
	if (setsockopt(sk, SOL_TCP, TCP_REPAIR_QUEUE, &val, sizeof(val)))
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","setsockopt() TCP_REPAIR_QUEUE,TCP_SEND_QUEUE error,errno:%d,%s",errno,strerror(errno));
		return -1;
	}

	val = tcp->seq;
	if (setsockopt(sk, SOL_TCP, TCP_QUEUE_SEQ, &val, sizeof(val)))
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","setsockopt() TCP_QUEUE_SEQ error,errno:%d,%s",errno,strerror(errno));
		return -1;
	}

	val = TCP_RECV_QUEUE;
	if (setsockopt(sk, SOL_TCP, TCP_REPAIR_QUEUE, &val, sizeof(val)))
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","setsockopt() TCP_REPAIR_QUEUE,TCP_RECV_QUEUE error,errno:%d,%s",errno,strerror(errno));
		return -1;
	}

	val = tcp->ack;
	if (setsockopt(sk, SOL_TCP, TCP_QUEUE_SEQ, &val, sizeof(val)))
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","setsockopt() TCP_QUEUE_SEQ error,errno:%d,%s",errno,strerror(errno));
		return -1;
	}


	/* ============= Bind and connect ================ */
	if(addr_type == ADDR_TYPE_IPV4)
	{
		if (bind(sk,client_addr, sizeof(struct sockaddr)))
		{
			MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","bind()	error,errno:%d,%s",errno,strerror(errno));
			return -1;
		}
		
		if (connect(sk,server_addr, sizeof(struct sockaddr)))
		{
			MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","connect()  error,errno:%d,%s",errno,strerror(errno));
			return -1;
		}

	}
	else if (addr_type == ADDR_TYPE_IPV6)
	{
		if (bind(sk,client_addr, sizeof(struct sockaddr_in6)))
		{
			MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","bind()	error,errno:%d,%s",errno,strerror(errno));
			return -1;
		}
		
		if (connect(sk,server_addr, sizeof(struct sockaddr_in6)))
		{
			MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","connect()  error,errno:%d,%s",errno,strerror(errno));
			return -1;
		}
	}


	if((tcp->wscale_dst)&&(tcp->wscale_src))
	{
		opts[onr].opt_code = TCPOPT_WINDOW;
		opts[onr].opt_val = tcp->wscale_dst+ (tcp->wscale_src<< 16);
		onr++;
	}

	
	opts[onr].opt_code = TCPOPT_MAXSEG;
	opts[onr].opt_val = (tcp->mss_src<tcp->mss_dst)?tcp->mss_src:tcp->mss_dst;
	onr++;

	if((tcp->sack_src)&&(tcp->sack_dst))
	{
		opts[onr].opt_code = TCPOPT_SACK_PERMITTED;
		opts[onr].opt_val = 0;
		onr++;
	}


//test
	
	struct sockaddr_in* client_addr_v4 = (struct sockaddr_in*)client_addr;
	struct sockaddr_in* server_addr_v4 = (struct sockaddr_in*)server_addr;
	if(addr_type == ADDR_TYPE_IPV4)
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_DEBUG,"tcprepair_set_option","sip:%d,dip:%d,sport:%d,dport:%d,wscale:%d,mss:%d",
					ntohl(client_addr_v4->sin_addr.s_addr),ntohl(server_addr_v4->sin_addr.s_addr),ntohs(client_addr_v4->sin_port),ntohs(server_addr_v4->sin_port),opts[0].opt_val,opts[1].opt_val);
	}

//test end

	if (setsockopt(sk, SOL_TCP, TCP_REPAIR_OPTIONS,opts, onr * sizeof(struct tcp_repair_opt)) < 0) 
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","setsockopt() TCP_REPAIR_OPTIONS error,errno:%d,%s",errno,strerror(errno));
		return -1;
	}

	val = 0;
	if (setsockopt(sk, SOL_TCP, TCP_REPAIR, &val, sizeof(val)))
	{
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","setsockopt() TCP_REPAIR close error,errno:%d,%s",errno,strerror(errno));
		return -1;
	}

	return 0;


}




int tcp_repair_process(const struct streaminfo* pstream,const void* a_packet,struct kni_pme_info* pmeinfo,int protocol)
{
	kni_filestate2_set(pstream->threadnum,COLUME_TCPREPAIR_TOTAL,0,2);

	int ret=0;
	struct kni_repaired_fds repaired_fds;

	int fd_client,fd_server;
	struct kni_tcp_state fake_client;
	struct kni_tcp_state fake_server;

	struct sockaddr* client_addr = NULL;
	struct sockaddr* server_addr = NULL;

	struct timespec start, end;
	long elapse = 0;

	
	if(pstream->addr.addrtype==ADDR_TYPE_IPV4)
	{
		fd_client = socket(AF_INET, SOCK_STREAM, 0);
		fd_server = socket(AF_INET, SOCK_STREAM, 0);
	}
	else if(pstream->addr.addrtype==ADDR_TYPE_IPV6)
	{
		fd_client = socket(AF_INET6, SOCK_STREAM, 0);
		fd_server = socket(AF_INET6, SOCK_STREAM, 0);
	}
	else
	{
		assert(0);
	}
	if ((fd_client < 0)||(fd_server<0))
	{
		kni_filestate2_set(pstream->threadnum,COLUME_TCPREPAIR_SOCKET_ERR,0,2);
		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","socket() error");
		return -1;
	}


	tcprepair_get_addr((void**)&client_addr,(void**)&server_addr,&(pstream->addr),a_packet);
	tcprepair_get_state(pstream->curdir,&(pstream->addr),&fake_client,&fake_server,(void*)a_packet,pmeinfo);
	
		
	ret=tcprepair_set_state(fd_client,&fake_server,server_addr,client_addr,pstream->addr.addrtype);
	if(ret<0)
	{
		kni_filestate2_set(pstream->threadnum,COLUME_TCPREPAIR_ERROR,0,2);
		close(fd_client);
		close(fd_server);
	
		tcprepair_free_addr(client_addr,server_addr);
//		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","fd_client tcprepair_set_state() error,dropme and fwdpkt");
		kni_log_debug(RLOG_LV_FATAL,(char*)"tcprepair_set_state",a_packet,(const char*)"fd_client tcprepair_set_state() error,dropme and fwdpkt");
		return -1;
	}
	
	ret=tcprepair_set_state(fd_server,&fake_client,client_addr,server_addr,pstream->addr.addrtype);
	if(ret<0)
	{
		kni_filestate2_set(pstream->threadnum,COLUME_TCPREPAIR_ERROR,0,2);
		close(fd_client);
		close(fd_server);
		
		tcprepair_free_addr(client_addr,server_addr);
//		MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,"tcprepair_set_state","fd_server tcprepair_set_state() error,dropme and fwdpkt");
		kni_log_debug(RLOG_LV_FATAL,(char*)"tcprepair_set_state",a_packet,(const char*)"fd_server tcprepair_set_state() error,dropme and fwdpkt");
		return -1;
	}

	tcprepair_free_addr(client_addr,server_addr);

	repaired_fds.client_fd = fd_client;
	repaired_fds.server_fd = fd_server;
	repaired_fds.protocol = pmeinfo->protocol;
	repaired_fds.keyring = pmeinfo->keyring_id;

	if(g_kni_switch_info.send_fds_mode == 0)
	{
		ret=MESA_lqueue_join_tail(g_kni_structinfo.lqueue_send_fds,(void*)&repaired_fds,sizeof(repaired_fds));
		if(ret <0)
	    {
			kni_filestate2_set(pstream->threadnum,COLUME_TCPREPAIR_JOINLQ_ERR,0,2);
		
			close(fd_client);
			close(fd_server);
		
			MESA_handle_runtime_log(g_kni_comminfo.logger, RLOG_LV_FATAL,KNI_MODULE_SENDFD,"MESA_lqueue_try_join_head() error,ret:%d",ret);
	    }
	}
	else
	{
		clock_gettime(CLOCK_MONOTONIC, &start);

		ret=kni_send_fds(g_kni_comminfo.fd_domain,repaired_fds);
		if(ret<0)	
		{
			kni_filestate2_set(0,COLUME_TCPREPAIR_SEND_ERR,0,2);
			g_kni_comminfo.kni_mode_cur=KNI_MODE_BYPASS;
		}
		else
		{
			kni_filestate2_set(0,COLUME_TCPREPAIR_SEND_SUCC,0,2);
		}
		
		close(fd_client);
		close(fd_server);	
		clock_gettime(CLOCK_MONOTONIC, &end);
							
		elapse=(end.tv_sec-start.tv_sec)*1000000+(end.tv_nsec-start.tv_nsec)/1000;
		FS_operate(g_kni_fs2_info.handler, g_kni_fs2_info.metric_sendfd, 0, FS_OP_SET, elapse);
		
	}
	
	pmeinfo->client_fd = fd_client;
	pmeinfo->server_fd = fd_server;

	return ret;

}