path: root/entry/src/kni_entry.cpp

/*
intercept:  destroy_pme + send_log + del traceid2pem + del tuple2stream

bypass: drome:  pme_new_fail:    destroy_pme
				no_tfe:          destroy_pme
			    intercept_error:    destroy_pme + send_log
		
		giveme: policy:          destroy_pme + send_log
				dup_traffic:     destroy_pme + send_log
*/
#define __STDC_FORMAT_MACROS
#include "kni_utils.h"
#include "marsio.h"
#include "MESA/stream_inc/sapp_inject.h"
#include "kni_cmsg.h"
#include <linux/if_ether.h>
#include <signal.h>
#include <inttypes.h>
#include "tfe_mgr.h"
#include <tsg/tsg_rule.h>
#include <tsg/tsg_send_log.h>

#include <stdio.h>
#include <errno.h>
#include <string.h>

#ifdef __cplusplus
extern "C" {
#endif
	#include "dablooms.h"
#ifdef __cplusplus
}
#endif
#include "kni_tap_rss.h"
#include <linux/if_tun.h>
#include <tsg/tsg_statistic.h>
#include <MESA/stream_inc/stream_control.h>
#include "kni_entry.h"
#include "kni_pxy_tcp_option.h"
#include "kni_dynamic_bypass.h"


#define KNI_VAR_VERSION 20_09_v20_09_a59d3a1

#ifdef __cplusplus
extern "C"
{
#endif

#define GIT_VERSION_CATTER(v) __attribute__((__used__)) const char * GIT_VERSION_##v = NULL
#define GIT_VERSION_EXPEND(v) GIT_VERSION_CATTER(v)

/* VERSION TAG */
#ifdef KNI_VAR_VERSION
GIT_VERSION_EXPEND(KNI_VAR_VERSION);
#else
static __attribute__((__used__)) const char * GIT_VERSION_UNKNOWN = NULL;
#endif
#undef GIT_VERSION_CATTER
#undef GIT_VERSION_EXPEND

#ifdef __cplusplus
}
#endif


struct kni_handle *g_kni_handle = NULL;
struct kni_field_stat_handle *g_kni_fs_handle = NULL;
int *arr_last_tfe_dispatch_index = NULL;

static int io_uring_add_write(int thread_seq, char *raw_data, uint16_t raw_len, addr_type_t addr_type);
static int io_uring_add_read(int thread_seq);
static int io_uring_polling(int thread_seq);

static char* stream_errmsg_session_record(enum intercept_error _errno){
	switch(_errno){
		case INTERCEPT_ERROR_ASYM_ROUTING:
			return (char*)"e_asym_routing";
		case INTERCEPT_ERROR_NO_SYN:
			return (char*)"e_no_syn";
		case INTERCEPT_ERROR_NO_SYN_ACK:
			return (char*)"e_no_synack";
		case INTERCEPT_ERROR_INVALID_IP_HDR:
			return (char*)"e_invalid_ip_hdr";
		case INTERCEPT_ERROR_EXCEED_MTU:
			return (char*)"e_exceed_mtu";
		case INTERCEPT_ERROR_SENDTO_TFE_FAIL:
			return (char*)"e_internal_1";
		case INTERCEPT_ERROR_TUPLE2STM_ADD_FAIL:
			return (char*)"e_internal_2";
		case INTERCEPT_ERROR_NO_TFE:
			return (char*)"e_internal_3";
		case INTERCEPT_ERROR_DUP_TRAFFIC:
			return (char*)"e_internal_4";
		case INTERCEPT_ERROR_CMSG_ADD_FAIL:
			return (char*)"e_internal_5";
		case INTERCEPT_ERROR_NOT_TCP_LINK_BYSYN:
			return (char*)"e_internal_6";
		case INTERCEPT_ERROR_GET_TCP_LINK_MODE_ERR:
			return (char*)"e_internal_7";
		case INTERCEPT_ERROR_STREAM_TUNNLE_TYPE:
			return (char *)"e_stream_type_tunnel";
		case INTERCEPT_ERROR_GET_STREAM_TUNNLE_TYPE_ERR:
			return (char *)"e_internal_8";
		case INTERCEPT_ERROR_GET_HAVE_DUP_PKT_ERR:
			return (char *)"e_internal_9";
		case INTERCEPT_ERROR_DUP_PKT_NOT_SURE_ERR:
			return (char *)"e_internal_10";
		default:
			return (char*)"unknown error";
	}
}

static int dup_traffic_dabloom_key_get(struct pkt_info *pktinfo, struct dup_traffic_dabloom_key *key){
	//ipv6
	struct tcphdr *tcphdr = pktinfo->tcphdr;
	key->seq = tcphdr->seq;
	key->ack_seq = tcphdr->ack_seq;
	struct kni_tcpopt_info tcpopt;
	kni_get_tcpopt(&tcpopt, tcphdr, pktinfo->tcphdr_len);
	key->timestamp = tcpopt.ts_value;
	if(pktinfo->addr_type == ADDR_TYPE_IPV6){
		struct ip6_hdr *iphdr = pktinfo->iphdr.v6;
		memcpy(key->addr.v6.saddr, &(iphdr->ip6_src), sizeof(key->addr.v6.saddr));
		memcpy(key->addr.v6.daddr, &(iphdr->ip6_dst), sizeof(key->addr.v6.daddr));
		key->addr.v6.source = tcphdr->source;
		key->addr.v6.dest = tcphdr->dest;
	}
	//ipv4
	else{
		struct iphdr *iphdr = pktinfo->iphdr.v4;
		key->addr.v4.saddr = iphdr->saddr;
		key->addr.v4.daddr = iphdr->daddr;
		key->addr.v4.source = tcphdr->source;
		key->addr.v4.dest = tcphdr->dest;
		key->ipid = iphdr->id;
	}
	return 0;
}

static void pme_info_destroy(void *data){
	struct pme_info *pmeinfo = (struct pme_info *)data;
	void *logger = g_kni_handle->local_logger;
	if(pmeinfo != NULL){
		//free lock
		pthread_mutex_destroy(&(pmeinfo->lock));
		//free syn/syn_ack
		FREE(&(pmeinfo->syn_packet));
		FREE(&(pmeinfo->syn_ack_packet));

		FREE(&pmeinfo);
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_PME_FREE], 0, FS_OP_ADD, 1);
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_PME_CNT], 0, FS_OP_ADD, -1);
	}
	else{
		KNI_LOG_ERROR(logger, "Failed at pme_info_destroy, pmeinfo = null");
	}
}

static int pme_info_init(struct pme_info *pmeinfo, const struct streaminfo *stream, int thread_seq){
	void *logger = g_kni_handle->local_logger;
	pmeinfo->stream = stream;
	pmeinfo->addr_type = (enum addr_type_t)stream->addr.addrtype;
	pmeinfo->ssl_cert_verify = -1;
	uint64_t traceid = tsg_get_stream_trace_id((struct streaminfo*)stream);
	snprintf(pmeinfo->stream_traceid, sizeof(pmeinfo->stream_traceid), "%" PRIu64 , traceid);
	if(g_kni_handle->log_level < RLOG_LV_FATAL)
		printaddr_r(&(stream->addr), pmeinfo->stream_addr, KNI_ADDR_MAX);
	//init pme_lock
	int ret = pthread_mutex_init(&(pmeinfo->lock), NULL);
	if(ret < 0){
		KNI_LOG_ERROR(logger, "Failed at init pthread mutex, stream traceid = %s, stream addr = %s", pmeinfo->stream_traceid, pmeinfo->stream_addr);
		abort();
	}
	return 0;
}

/*keys: 
common: common_has_dup_traffic, common_stream_error
http: http_host
ssl: ssl_sni, ssl_pinningst, ssl_intercept_state, ssl_server_side_latency, ssl_client_side_latency, ssl_server_side_version, ssl_client_side_version, 
	 ssl_cert_verify
*/
static int log_generate(struct pme_info *pmeinfo){		
	void *local_logger = g_kni_handle->local_logger;
	struct TLD_handle_t *tld_handle = pmeinfo->tld_handle;
	//common
	//schema_type
	TLD_append(tld_handle, (char*)"common_schema_type", (void*)(pmeinfo->protocol == PROTO_SSL ? "SSL" : "HTTP"),  TLD_TYPE_STRING);
	//dup_traffic
	TLD_append(tld_handle, (char*)"common_has_dup_traffic", (void*)pmeinfo->has_dup_traffic, TLD_TYPE_LONG);
	//intercept_error
	if(pmeinfo->intcp_error < 0){
		char *stream_errmsg = stream_errmsg_session_record(pmeinfo->intcp_error);
		TLD_append(tld_handle, (char*)"common_stream_error", (void*)stream_errmsg, TLD_TYPE_STRING);
	}

	//ssl
	if(pmeinfo->protocol == PROTO_SSL){
		TLD_append(tld_handle, (char*)"ssl_sni", (void*)pmeinfo->domain.sni, TLD_TYPE_STRING);
		//pinning state: from tfe
		TLD_append(tld_handle, (char*)"intercept_pinning_status", (void*)(long)pmeinfo->ssl_pinningst, TLD_TYPE_LONG);
		//intercept state: from tfe
		TLD_append(tld_handle, (char*)"intercept_status", (void*)(long)pmeinfo->ssl_intercept_state, TLD_TYPE_LONG);
		//ssl upstream latency: from tfe
		TLD_append(tld_handle, (char*)"intercept_server_side_latency", (void*)pmeinfo->ssl_server_side_latency, TLD_TYPE_LONG);
		//ssl downstream latency: from tfe
		TLD_append(tld_handle, (char*)"intercept_client_side_latency", (void*)pmeinfo->ssl_client_side_latency, TLD_TYPE_LONG);
		//ssl upstream version: from tfe
		TLD_append(tld_handle, (char*)"intercept_server_side_version", (void*)pmeinfo->ssl_server_side_version, TLD_TYPE_STRING);
		//ssl downstream version: from tfe
		TLD_append(tld_handle, (char*)"intercept_client_side_version", (void*)pmeinfo->ssl_client_side_version, TLD_TYPE_STRING);
		//ssl error: from tfe
		if(strlen(pmeinfo->ssl_error) > 0)
			TLD_append(tld_handle, (char*)"intercept_error", (void*)pmeinfo->ssl_error, TLD_TYPE_STRING);
		//ssl cert verify
		if(pmeinfo->ssl_cert_verify != -1){
			TLD_append(tld_handle, (char*)"intercept_cert_verify", (void*)(long)pmeinfo->ssl_cert_verify, TLD_TYPE_LONG);
		}
		if (pmeinfo->ssl_intercept_state == 0){
			if (strlen(pmeinfo->ssl_passthrough_reason))
			{
				TLD_append(tld_handle, (char*)"intercept_passthrough_reason", (void*)pmeinfo->ssl_passthrough_reason, TLD_TYPE_STRING);
			}
			else if (pmeinfo->intcp_error < 0)
			{
				TLD_append(tld_handle, (char*)"intercept_passthrough_reason", (void*)"Underlying Stream Error", TLD_TYPE_STRING);
			}
			else
			{
				TLD_append(tld_handle, (char*)"intercept_passthrough_reason", (void*)"Unknown", TLD_TYPE_STRING);
			}
		}
	}
	//host
	if(pmeinfo->protocol == PROTO_HTTP){
		TLD_append(tld_handle, (char*)"http_host", (void*)pmeinfo->domain.host, TLD_TYPE_STRING);
	}
	//int ret = tsg_send_log(g_tsg_log_instance, tld_handle, NULL, &pmeinfo->maat_rule, (size_t)pmeinfo->n_maat_rule, 0);

	int ret = tsg_send_log(g_tsg_log_instance, tld_handle, NULL, LOG_TYPE_INTERCEPT_EVENT, &pmeinfo->maat_rule, 1, 0);
	//int ret = tsg_send_log(g_tsg_log_instance, tld_handle, &log_msg, 0);
	if(ret < 0){
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_SENDLOG_FAIL], 0, FS_OP_ADD, 1);
		KNI_LOG_ERROR(local_logger, "Failed at sendlog, ret = %d, strem_traceid = %s", 
				ret, pmeinfo->stream_traceid);
		goto error_out;
	}
	if(pmeinfo->protocol == PROTO_SSL){
		switch(pmeinfo->ssl_pinningst)
		{
			case 0:
				tsg_set_statistic_opt(1,OPT_TYPE_PINNING_NOT, 0);
				break;
			case 1:
				tsg_set_statistic_opt(1,OPT_TYPE_PINNING_YES, 0);
				break;
			case 2:
				tsg_set_statistic_opt(1,OPT_TYPE_PINNING_MAYBE, 0);
				break;
			default:
				break;
		}
	}
	FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_SENDLOG_SUCC], 0, FS_OP_ADD, 1);
	return 0;

error_out:
	return -1;
}

static int log_generate_no_intercept(struct pme_info *pmeinfo)
{
    int ret = 0;
    void *local_logger = g_kni_handle->local_logger;
    struct TLD_handle_t *tld_handle = pmeinfo->tld_handle;

    switch(pmeinfo->protocol)
    {
        case PROTO_SSL:
            TLD_append(tld_handle, (char*)"ssl_sni", (void*)pmeinfo->domain.sni, 
                       TLD_TYPE_STRING);

            TLD_append(tld_handle, (char*)"common_schema_type", (void*)"SSL", 
                       TLD_TYPE_STRING);
            break;
        case PROTO_HTTP:
            TLD_append(tld_handle, (char*)"http_host", (void*)pmeinfo->domain.host,
                       TLD_TYPE_STRING);
            TLD_append(tld_handle, (char*)"common_schema_type", (void*)"HTTP", 
                       TLD_TYPE_STRING);
            break;
        default:
            break;
    }

    ret = tsg_send_log(g_tsg_log_instance, tld_handle, NULL,
                       LOG_TYPE_INTERCEPT_EVENT, &pmeinfo->maat_rule, 1, 0);
    if(ret < 0)
    {
        FS_operate(g_kni_fs_handle->handle, 
                   g_kni_fs_handle->fields[KNI_FIELD_SENDLOG_FAIL],
                   0, FS_OP_ADD, 1);
        KNI_LOG_ERROR(local_logger, "Failed at sendlog, ret = %d, "
                                    "strem_traceid = %s", 
                      ret, pmeinfo->stream_traceid);

        return -1;
    }

    FS_operate(g_kni_fs_handle->handle, 
               g_kni_fs_handle->fields[KNI_FIELD_SENDLOG_SUCC], 
               0, FS_OP_ADD, 1);
    return 0;
}



static void set_proxy_rule_hits_metric(struct pme_info *pmeinfo, int thread_id)
{
    void *logger = g_kni_handle->local_logger; 
    struct proxy_metric_value *metric_value = &(pmeinfo->proxy_metric_value);
    struct proxy_metric_tag metric_tag;
    memset(&metric_tag, 0, sizeof(struct proxy_metric_tag));

    metric_tag.action = pmeinfo->maat_rule.action;
    metric_tag.rule_id = pmeinfo->maat_rule.rule_id;
    metric_tag.vsys_id = pmeinfo->maat_rule.vsys_id;
    metric_tag.pinning_status = pmeinfo->ssl_intercept_state;

    proxy_set_metric_value(g_kni_handle->proxy_fieldstat, &metric_tag,
                            metric_value, thread_id);

    KNI_LOG_DEBUG(logger, "Set proxy_rule_hits metric, action = %d, "
                          "rule_id = %lu, vsys_id = %d, pinning_status = %d, "
                          "hit_count = %d, in_bytes = %d, out_bytes = %d, "
                          "in_pkts = %d, out_pkts = %d, stream_id = %s",
                  metric_tag.action, metric_tag.rule_id, metric_tag.vsys_id,
                  metric_tag.pinning_status, metric_value->hit_count,
                  metric_value->in_bytes, metric_value->in_pkts, 
                  metric_value->out_bytes, metric_value->out_pkts, 
                  pmeinfo->stream_traceid);

    return;
}


static void send_log_message(struct pme_info *pmeinfo)
{
    int ret = 0;
    void *logger = g_kni_handle->local_logger;

    switch(pmeinfo->action)
    {
        case KNI_ACTION_INTERCEPT:
            ret = log_generate(pmeinfo);
            break;
        case KNI_ACTION_NO_INTERCEPT:
            ret = log_generate_no_intercept(pmeinfo);
            break;
        default:
            return;
            break;
    }

    if(ret < 0)
    {
        KNI_LOG_ERROR(logger, "Failed at log_generate, stream traceid = %s,"
                              " stream addr = %s", 
                      pmeinfo->stream_traceid, pmeinfo->stream_addr);
    }
    else
    {
        KNI_LOG_DEBUG(logger, "Succeed at log_generate, stream traceid = %s,"
                              " stream addr = %s",
                      pmeinfo->stream_traceid, pmeinfo->stream_addr);
    }
    return;
}

static void stream_destroy(struct pme_info *pmeinfo)
{
    //sendlog
    send_log_message(pmeinfo);
    //free pme
    pme_info_destroy(pmeinfo);
}

static int judge_stream_can_destroy(struct pme_info *pmeinfo, int caller){
	void *logger = g_kni_handle->local_logger;
	int can_destroy = 0;
	if(pmeinfo != NULL){
		pthread_mutex_lock(&(pmeinfo->lock));
		if(caller == CALLER_SAPP){
			pmeinfo->sapp_release = 1;
		}
		if(caller == CALLER_TFE){
			pmeinfo->tfe_release = 1;
		}
		if(pmeinfo->sapp_release == 1 && pmeinfo->tfe_release == 1){
			can_destroy = 1;
		}
		pthread_mutex_unlock(&(pmeinfo->lock));
	}
	else{
		KNI_LOG_ERROR(logger, "Failed at judge_stream_can_destroy, pmeinfo = null");
	}
	return can_destroy;
}

int wrapped_kni_cmsg_set(struct kni_cmsg *cmsg, uint16_t type, const unsigned char *value, uint16_t size, struct pme_info *pmeinfo){
	void *logger = g_kni_handle->local_logger;
	int ret = kni_cmsg_set(cmsg, type, value, size);
	if(ret < 0){
		KNI_LOG_ERROR(logger, "Failed set cmsg, type = %d/%s, stream traceid = %s, stream addr = %s", type, tfe_cmsg_tlv_type_to_string[type],pmeinfo->stream_traceid, pmeinfo->stream_addr);
	}
	else
	{
		KNI_LOG_DEBUG(logger, "Successd to set cmsg, type = %d/%s, stream traceid = %s, stream addr = %s", type,tfe_cmsg_tlv_type_to_string[type], pmeinfo->stream_traceid, pmeinfo->stream_addr);
	}

	return ret;
}


static int session_attr_subscribe_id_cmsg_set(const struct session_runtime_attribute *session_attr, struct kni_cmsg *cmsg, struct pme_info *pmeinfo)
{
	int ret = 0;
	char empty_arr[MAX_STR_FIELD_LEN] = {0};
	const char *client_subscribe_id = srt_attribute_get_client_subscriber_id(session_attr);
	const char *server_subscribe_id = srt_attribute_get_server_subscriber_id(session_attr);

	do{
		if(client_subscribe_id == NULL)
		{
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_SUB_ID, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo);
		}
		else
		{
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_SUB_ID, (const unsigned char*)client_subscribe_id, strlen(client_subscribe_id), pmeinfo);
		}
		if(ret < 0) break;

		if(server_subscribe_id == NULL)
		{
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_SUB_ID, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo);
		}
		else
		{
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_SUB_ID, (const unsigned char*)server_subscribe_id, strlen(server_subscribe_id), pmeinfo);	
		}
		if(ret < 0) break;
	}while(0);

	return ret;
}

static int session_attr_asn_cmsg_set(const struct session_runtime_attribute *session_attr, struct kni_cmsg *cmsg, struct pme_info *pmeinfo)
{
	int ret = 0;
	char empty_arr[MAX_STR_FIELD_LEN] = {0};
	const struct asn_info *client_asn = srt_attribute_get_client_ip_asn(session_attr);
	const struct asn_info *server_asn = srt_attribute_get_server_ip_asn(session_attr);

	do{
		if(client_asn == NULL)
		{
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_ASN, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo);
			if(ret < 0) break;
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_ORGANIZATION, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo);
			if(ret < 0) break;
		}
		else
		{
			client_asn->asn_id == NULL
				?ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_ASN, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo)
				:ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_ASN, (const unsigned char*)client_asn->asn_id, strlen(client_asn->asn_id), pmeinfo);

			if(ret < 0) break;
			
			client_asn->organization == NULL
				?ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_ORGANIZATION, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo)
				:ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_ORGANIZATION, (const unsigned char*)client_asn->organization, strlen(client_asn->organization), pmeinfo);
			if(ret < 0) break;
		}

		if(server_asn == NULL)
		{
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_ASN, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo);
			if(ret < 0) break;
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_ORGANIZATION, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo);
			if(ret < 0) break;
		}
		else
		{
			server_asn->asn_id == NULL
				?ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_ASN, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo)
				:ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_ASN, (const unsigned char*)server_asn->asn_id, strlen(server_asn->asn_id), pmeinfo);
			if(ret < 0) break;

			server_asn->organization == NULL
				?ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_ORGANIZATION, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo)
				:ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_ORGANIZATION, (const unsigned char*)server_asn->organization, strlen(server_asn->organization), pmeinfo);
			if(ret < 0) break;
		}
	}while(0);
	return ret;
}

static int session_attr_ip_location_cmsg_set(const struct session_runtime_attribute *session_attr, struct kni_cmsg *cmsg, struct pme_info *pmeinfo)
{
	int ret = 0;
	char empty_arr[MAX_STR_FIELD_LEN] = {0};
	const struct location_info *client_location = srt_attribute_get_client_ip_location(session_attr);
	const struct location_info *server_location = srt_attribute_get_server_ip_location(session_attr);

	do{
		if(client_location == NULL || client_location->full_location == NULL)
		{
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_IP_LOCATION, (const unsigned char*)empty_arr, 0, pmeinfo);
			if(ret < 0) break;
		}
		else
		{
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_IP_LOCATION, (const unsigned char*)client_location->full_location, (uint16_t)client_location->full_location_len, pmeinfo);
			if(ret < 0) break;
		}

		if(server_location == NULL || server_location->full_location == NULL)
		{
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_IP_LOCATION, (const unsigned char*)empty_arr, 0, pmeinfo);
			if(ret < 0) break;
		}
		else
		{
			ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_IP_LOCATION, (const unsigned char*)server_location->full_location, (uint16_t)server_location->full_location_len, pmeinfo);
			if(ret < 0) break;
		}
	}while(0);
	return ret;
}


static int session_attr_ja3_cmsg_set(const struct session_runtime_attribute *session_attr, struct kni_cmsg *cmsg, struct pme_info *pmeinfo)
{
	int ret = 0;
	char empty_arr[MAX_STR_FIELD_LEN] = {0};

	const char *ja3_fingerprint = srt_attribute_get_ja3_fingerprint(session_attr);

	ja3_fingerprint == NULL
		?ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SSL_CLIENT_JA3_FINGERPRINT, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo)
		:ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SSL_CLIENT_JA3_FINGERPRINT, (const unsigned char*)ja3_fingerprint, strlen(ja3_fingerprint), pmeinfo);

	return ret;
}



static int session_attr_fqdn_category_id_cmsg_set(const struct session_runtime_attribute *session_attr, struct kni_cmsg *cmsg, struct pme_info *pmeinfo)
{
	int ret = 0;
	size_t n_category_ids = 0;
	unsigned int category_ids[8] = {0};
	char fqdn_val[sizeof(int) * 8] = {0};
	char empty_arr[MAX_STR_FIELD_LEN] = {0};

	n_category_ids = srt_attribute_get_category_ids((const struct session_runtime_attribute *)pmeinfo->session_attribute, category_ids, sizeof(category_ids)/sizeof(category_ids[0]));

	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_FQDN_CAT_ID_NUM, (const unsigned char*)&n_category_ids, sizeof(unsigned int), pmeinfo);
	if(n_category_ids <= 0 || n_category_ids > 8)
	{
		ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_FQDN_CAT_ID_VAL, (const unsigned char*)empty_arr, strlen(empty_arr), pmeinfo);
	}
	else
	{
		for(int i = 0 ; i < (int)n_category_ids; i++)
		{
			memcpy((void *)(fqdn_val + i * (sizeof(unsigned int))),  (void *)&(category_ids[i]), sizeof(unsigned int));
		}
		ret = wrapped_kni_cmsg_set(cmsg ,TFE_CMSG_FQDN_CAT_ID_VAL, (const unsigned char*)fqdn_val, n_category_ids * sizeof(unsigned int) , pmeinfo);
	}
	return ret;
}



static int session_attribute_cmsg_set(struct kni_cmsg *cmsg, struct pme_info *pmeinfo)
{
	int ret = 0;
	struct session_runtime_attribute *session_attr = pmeinfo->session_attribute;

	if(session_attr == NULL)
	{
		return -1;
	}
	do {
		ret = session_attr_subscribe_id_cmsg_set(session_attr, cmsg, pmeinfo);
		if(ret < 0) break;

		ret = session_attr_asn_cmsg_set(session_attr, cmsg, pmeinfo);
		if(ret < 0) break;

		ret = session_attr_ip_location_cmsg_set(session_attr, cmsg, pmeinfo);
		if(ret < 0) break;

		ret = session_attr_ja3_cmsg_set(session_attr, cmsg, pmeinfo);
		if(ret < 0) break;

		ret = session_attr_fqdn_category_id_cmsg_set(session_attr, cmsg, pmeinfo);
		if(ret < 0) break;
	}while(0);
	return ret;
}


static unsigned int get_stream_common_direction(struct streaminfo *stream)
{
	int i_or_e=0;
	unsigned int direction=0;
	i_or_e=MESA_dir_link_to_human(stream->routedir);
	switch(stream->curdir)
	{
		case	DIR_C2S:
			if(i_or_e=='E' || i_or_e=='e')
			{
				direction='E';
			}
			else
			{
				direction='I';
			}
			break;
		case	DIR_S2C:
			if(i_or_e=='E' || i_or_e=='e')
			{
				direction='I';
			}
			else
			{
				direction='E';
			}
			break;
		default:
			break;
	}
	return direction;
}

static unsigned char* kni_cmsg_serialize_header_new(struct pme_info *pmeinfo, struct streaminfo *stream, struct pkt_info *pktinfo, uint16_t *len){
	void *logger = g_kni_handle->local_logger;
	uint16_t bufflen = 0, serialize_len = 0;
	unsigned char *buff = NULL;
	//uint8_t protocol_type = pmeinfo->protocol == PROTO_SSL ? 0x1 : 0x0;
	uint8_t protocol_type = 0x0;	
	struct kni_cmsg *cmsg = kni_cmsg_init();
	char *trace_id = NULL;
	uint32_t seq = pktinfo->tcphdr->seq;
	uint32_t ack = pktinfo->tcphdr->ack_seq;
	if(g_kni_handle->pxy_tcp_option_enable == 1){
		pxy_tcp_option_modify_mss(pmeinfo, logger);
	}
	uint16_t client_mss = htons(pmeinfo->client_tcpopt.mss);
	uint16_t server_mss = htons(pmeinfo->server_tcpopt.mss);
	uint16_t client_window = htons(pmeinfo->client_window);
	uint16_t server_window = htons(pmeinfo->server_window);
	unsigned char stream_curdir = stream->curdir;	
	if(stream_curdir == DIR_S2C)
	{
		seq = pktinfo->tcphdr->ack_seq;
		ack = pktinfo->tcphdr->seq;
	}

	char src_mac[6] = {0};
	char dst_mac[6] = {0};
	uint64_t policy_id;
	unsigned int stream_common_direction;
	
	switch(pmeinfo->protocol)
	{
		case PROTO_SSL:
		     protocol_type = 0x1;
		     break;
		
		case PROTO_SSH:
		     protocol_type = 0x2;
		     break;
		
		default:
		     protocol_type = 0x0;

	}
	//seq
	int ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_SEQ, (const unsigned char*)&seq, 4, pmeinfo);
	if(ret < 0) goto error_out;
	//ack
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_ACK, (const unsigned char*)&ack, 4, pmeinfo);
	if(ret < 0) goto error_out;
	//client mss
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_MSS_CLIENT, (const unsigned char*)&client_mss, 2, pmeinfo);
	if(ret < 0) goto error_out;
	//server mss
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_MSS_SERVER, (const unsigned char*)&server_mss, 2, pmeinfo);
	if(ret < 0) goto error_out;
	//both = 1, send to tfe
	if(pmeinfo->client_tcpopt.wscale_set && pmeinfo->server_tcpopt.wscale_set){
		//client wscale
		ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_WSACLE_CLIENT, (const unsigned char*)&(pmeinfo->client_tcpopt.wscale), 1, pmeinfo);
		if(ret < 0) goto error_out;
		//server wscale
		ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_WSACLE_SERVER, (const unsigned char*)&(pmeinfo->server_tcpopt.wscale), 1, pmeinfo);
		if(ret < 0) goto error_out;
	}
	//client sack
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_SACK_CLIENT, (const unsigned char*)&(pmeinfo->client_tcpopt.sack), 1, pmeinfo);
	if(ret < 0) goto error_out;
	//server sack
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_SACK_SERVER, (const unsigned char*)&(pmeinfo->server_tcpopt.sack), 1, pmeinfo);
	if(ret < 0) goto error_out;
	//client timestamp
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_TS_CLIENT, (const unsigned char*)&(pmeinfo->client_tcpopt.ts_set), 1, pmeinfo);
	if(ret < 0) goto error_out;
	//server  timestamp
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_TS_SERVER, (const unsigned char*)&(pmeinfo->server_tcpopt.ts_set), 1, pmeinfo);
	if(ret < 0) goto error_out;
	//client timestamp val
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_TS_CLIENT_VAL, (const unsigned char*)&(pmeinfo->client_tcpopt.ts_value), 4, pmeinfo);
	if(ret < 0) goto error_out;
	//server timestamp val
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_TS_SERVER_VAL, (const unsigned char*)&(pmeinfo->server_tcpopt.ts_value), 4, pmeinfo);
	if(ret < 0) goto error_out;
	//protocol
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_PROTOCOL, (const unsigned char*)&protocol_type, 1, pmeinfo);
	if(ret < 0) goto error_out;
	//client window
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_WINDOW_CLIENT, (const unsigned char*)&client_window, 2, pmeinfo);
	if(ret < 0) goto error_out;
	//server window
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_WINDOW_SERVER, (const unsigned char*)&server_window, 2, pmeinfo);
	if(ret < 0) goto error_out;
	//current packet direction
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_TCP_RESTORE_INFO_PACKET_CUR_DIR, (const unsigned char*)&stream_curdir, 1, pmeinfo);
	if(ret < 0) goto error_out;
	//maat policy id 
	policy_id = (uint64_t)pmeinfo->policy_id;
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_POLICY_ID, (const unsigned char*)&policy_id, sizeof(policy_id), pmeinfo);
	if(ret < 0) goto error_out;
	//stream trace id
	trace_id = pmeinfo->stream_traceid;
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_STREAM_TRACE_ID, (const unsigned char*)trace_id, 
					strnlen(pmeinfo->stream_traceid, sizeof(pmeinfo->stream_traceid)), pmeinfo);
	if(ret < 0) goto error_out;
	// proxy tcp option
	if(g_kni_handle->pxy_tcp_option_enable == 1)
	{
		ret = pxy_tcp_option_cmsg_set(cmsg, pmeinfo);
		if(ret < 0) 
		{
			KNI_LOG_DEBUG(logger, "Proxy-tcp-option: Failed at set cmsg");
			goto error_out;
		}
	}

	//share session attribute
	ret = session_attribute_cmsg_set(cmsg, pmeinfo);
	if(ret < 0)
	{
		KNI_LOG_ERROR(logger, "share-session-attribute:  Failed at set cmsg");
		goto error_out;
	}

    //common direction
	stream_common_direction = get_stream_common_direction(stream);
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_COMMON_DIRECTION, (const unsigned char*)&stream_common_direction, sizeof(stream_common_direction), pmeinfo);
	if(ret < 0) goto error_out;

	//src mac
	ret = get_rawpkt_opt_from_streaminfo(stream, RAW_PKT_GET_ORIGINAL_LOWEST_ETH_SMAC, src_mac);
	if(ret < 0){
		KNI_LOG_DEBUG(logger, "Failed at get src mac from rawpkt, ret = %d, maybe two-arm mode", ret);
	}
	KNI_LOG_DEBUG(logger, "Succeed at get src mac from rawpkt, addr = %02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx", 
					src_mac[0], src_mac[1], src_mac[2], src_mac[3], src_mac[4], src_mac[5]);
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_SRC_MAC, (const unsigned char*)src_mac, sizeof(src_mac), pmeinfo);
	if(ret < 0) goto error_out;
	//dst mac
	ret = get_rawpkt_opt_from_streaminfo(stream, RAW_PKT_GET_ORIGINAL_LOWEST_ETH_DMAC, dst_mac);
	if(ret < 0){
		KNI_LOG_DEBUG(logger, "Failed at get dst mac from rawpkt, ret = %d, maybe two-arm mode", ret);
	}
	KNI_LOG_DEBUG(logger, "Succeed at get dst mac from rawpkt, addr = %02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx", 
					dst_mac[0], dst_mac[1], dst_mac[2], dst_mac[3], dst_mac[4], dst_mac[5]);
	ret = wrapped_kni_cmsg_set(cmsg, TFE_CMSG_DST_MAC, (const unsigned char*)dst_mac, sizeof(dst_mac), pmeinfo);
	if(ret < 0) goto error_out;
	
	bufflen = kni_cmsg_serialize_size_get(cmsg);
	KNI_LOG_DEBUG(logger, "Successd set cmsg size:%d, stream traceid = %s", bufflen, pmeinfo->stream_traceid);
	buff = (unsigned char*)ALLOC(char, bufflen);
	serialize_len = 0;
	ret = kni_cmsg_serialize(cmsg, buff, bufflen, &serialize_len);
	if(ret < 0){
		KNI_LOG_ERROR(logger, "Failed at serialize cmsg, ret = %d, stream traceid = %s, stream addr = %s", 
					ret, pmeinfo->stream_traceid, pmeinfo->stream_addr);
		goto error_out;
	} 
	*len = serialize_len;
	kni_cmsg_destroy(cmsg);
	return buff;

error_out:
	if(buff != NULL){
		FREE(&buff);
	}
	kni_cmsg_destroy(cmsg);
	return NULL;
}

static char* add_cmsg_to_packet(struct pme_info *pmeinfo, struct streaminfo *stream, struct pkt_info *pktinfo, int *len){
	//tcp option: kind 88, len 4, control_info_len
	void * logger = g_kni_handle->local_logger;
	char *new_pkt = (char*)ALLOC(struct wrapped_packet, 1);
	int offset = 0;
	//iphdr
	KNI_LOG_DEBUG(logger, "Kni add cmsg to packet malloc buffer size:%d",sizeof(struct wrapped_packet));
	if(pmeinfo->addr_type == ADDR_TYPE_IPV6){
		memcpy(new_pkt, (void*)pktinfo->iphdr.v6, pktinfo->iphdr_len);
	}
	else{
		memcpy(new_pkt, (void*)pktinfo->iphdr.v4, pktinfo->iphdr_len);
	}
	offset += pktinfo->iphdr_len;
	//tcphdr
	struct tcphdr *tcphdr = (struct tcphdr*)(new_pkt + offset);
	memcpy(new_pkt + offset, (void*)pktinfo->tcphdr, 20);
	offset += 20;
	tcphdr->doff = pktinfo->tcphdr->doff + 1;
	struct tcp_option_restore *opt = ALLOC(struct tcp_option_restore, 1);
	opt->kind = 88;
	opt->len = 4;
	opt->offset = htons(pktinfo->data_len);
	memcpy(new_pkt + offset, (void*)opt, 4);
	FREE(&opt);
	offset += 4;
	memcpy(new_pkt + offset, (void*)((char*)pktinfo->tcphdr + 20), pktinfo->tcphdr_len - 20);
	offset += pktinfo->tcphdr_len - 20;
	//data
	memcpy(new_pkt + offset, (void*)pktinfo->data, pktinfo->data_len);
	offset += pktinfo->data_len;
	//kni_cmsg_serialize_header
	uint16_t header_len = 0;
	unsigned char* header = kni_cmsg_serialize_header_new(pmeinfo, stream, pktinfo, &header_len);
	if(header == NULL){
		KNI_LOG_ERROR(logger, "Kni add cmsg to packet: serialize_header failed");
		goto error_out;
	}
	memcpy(new_pkt + offset, (void*)header, header_len);
	offset += header_len;
	KNI_LOG_DEBUG(logger, "Kni add cmsg to packet:offset=%d,header_len=%d, tcp_data_len=%d",offset,header_len,pktinfo->data_len);
	FREE(&header);
	//ipv6
	if(pmeinfo->addr_type == ADDR_TYPE_IPV6){
		kni_ipv6_header_parse((void*)new_pkt, pktinfo);
		pktinfo->iphdr.v6->ip6_ctlun.ip6_un1.ip6_un1_plen = htons(offset - sizeof(ip6_hdr));
		pktinfo->tcphdr->check = 0;
		pktinfo->tcphdr->check = kni_tcp_checksum_v6((void*)pktinfo->tcphdr, 
					offset - pktinfo->iphdr_len, pktinfo->iphdr.v6->ip6_src, pktinfo->iphdr.v6->ip6_dst);
	}
	else{
		struct iphdr *iphdr = (struct iphdr*)new_pkt;
		iphdr->tot_len = htons(offset);
		//must set check = 0
		iphdr->check = 0;
		iphdr->check = kni_ip_checksum((void*)iphdr, pktinfo->iphdr_len);
		//tcphdr: checkdum
		tcphdr->check = 0;
		tcphdr->check = kni_tcp_checksum((void*)tcphdr, offset - pktinfo->iphdr_len, iphdr->saddr, iphdr->daddr);	
	}
	*len = offset;
	return new_pkt;

error_out:
	if(new_pkt != NULL){
		FREE(&new_pkt);
	}
	return NULL;
}

static int add_ether_header(void *dst_data, void *raw_data, uint16_t raw_len, addr_type_t addr_type){
	char *src_mac = g_kni_handle->src_mac_addr;
	char *dst_mac = g_kni_handle->dst_mac_addr;
	//ether_header[14]
	struct ethhdr *ether_hdr = (struct ethhdr*)dst_data;
	memcpy(ether_hdr->h_dest, dst_mac, sizeof(ether_hdr->h_dest));
	memcpy(ether_hdr->h_source, src_mac, sizeof(ether_hdr->h_source));
	if(addr_type == ADDR_TYPE_IPV6){
		ether_hdr->h_proto = htons(ETH_P_IPV6);
	}
	else{
		ether_hdr->h_proto = htons(ETH_P_IP);
	}
	memcpy((char*)dst_data + sizeof(*ether_hdr), raw_data, raw_len);
	return 0;
}

static int send_to_tfe_normal_mode(char *raw_data, uint16_t raw_len, int thread_seq, int tfe_id, addr_type_t addr_type){
	void *logger = g_kni_handle->local_logger;
	struct kni_marsio_handle *handle = g_kni_handle->marsio_handle;
	marsio_buff_t *tx_buffs[BURST_MAX];
	int index = -1;
	for(int i = 0; i < handle->tfe_enabled_node_count; i++){
		if(handle->tfe_enabled_nodes[i].tfe_id == tfe_id){
			index = i;
			break;
		}
	}
	if(index == -1){
		KNI_LOG_ERROR(logger, "tfd %d = disabled");
		return -1;
	}
	struct mr_vdev *dev_eth_handler = handle->tfe_enabled_nodes[index].dev_eth_handler;
	struct mr_sendpath *dev_eth_sendpath = handle->tfe_enabled_nodes[index].dev_eth_sendpath;
	//only send one packet, alloc_ret <= nr_send <= BURST_MAX
	int nr_send = 1;
	int alloc_ret = marsio_buff_malloc_device(dev_eth_handler, tx_buffs, nr_send, 0, thread_seq);
    if (alloc_ret < 0){
		KNI_LOG_ERROR(logger, "Failed at alloc marsio buffer, ret = %d, thread_seq = %d", 
					alloc_ret, thread_seq);
		return -1;
	}
	for(int i = 0; i < nr_send; i++){
		char* dst_data = marsio_buff_append(tx_buffs[i], raw_len + sizeof(struct ethhdr));
		add_ether_header(dst_data, raw_data, raw_len, addr_type);
	}
	marsio_send_burst(dev_eth_sendpath, thread_seq, tx_buffs, nr_send);
	return 0;
}

static int send_to_tfe_tap_mode(char *raw_data, uint16_t raw_len, int thread_seq, addr_type_t addr_type, void *logger)
{
	if (g_kni_handle->iouring_conf.enable_iouring)
	{
		if (io_uring_add_write(thread_seq, raw_data, raw_len, addr_type) == -1)
		{
			return -1;
		}
	}
	else
	{
		int tap_fd = g_kni_handle->threads_handle[thread_seq].tap_fd;
		char *send_buff = g_kni_handle->threads_handle[thread_seq].buff;
		add_ether_header(send_buff, raw_data, raw_len, addr_type);
		int ret = kni_tap_write_per_thread(tap_fd, send_buff, raw_len + sizeof(struct ethhdr), logger);
		if (ret < 0)
		{
			return -1;
		}
	}

	return 0;
}

static int send_to_tfe(char *raw_data, uint16_t raw_len, int thread_seq, int tfe_id, addr_type_t addr_type, void *logger)
{
	int mode = g_kni_handle->deploy_mode;
	switch (mode)
	{
	case KNI_DEPLOY_MODE_TAP:
		return send_to_tfe_tap_mode(raw_data, raw_len, thread_seq, addr_type, logger);
	case KNI_DEPLOY_MODE_NORMAL:
		return send_to_tfe_normal_mode(raw_data, raw_len, thread_seq, tfe_id, addr_type);
	default:
		assert(0);
		return -1;
	}
}

static void wrapped_kni_header_parse(const void *a_packet, struct pme_info *pmeinfo, struct pkt_info *pktinfo){
	void *logger = g_kni_handle->local_logger;
	if(pmeinfo->addr_type == ADDR_TYPE_IPV6){
		int ret = kni_ipv6_header_parse(a_packet, pktinfo);
		if(ret < 0){
			char *errmsg = kni_ipv6_errmsg_get((enum kni_ipv6hdr_parse_error)ret);
			KNI_LOG_ERROR(logger, "Failed at parse ipv6 header, errmsg = %s, stream treaceid = %s", 
					errmsg, pmeinfo->stream_traceid);
			pktinfo->parse_failed = 1;
		}
	}
	else{
		int ret = kni_ipv4_header_parse(a_packet, pktinfo);
		if(ret < 0){
			char *errmsg = kni_ipv4_errmsg_get((enum kni_ipv4hdr_parse_error)ret);
			KNI_LOG_ERROR(logger, "Failed at parse ipv4 header, errmsg = %s, stream treaceid = %s", 
					errmsg, pmeinfo->stream_traceid);
			pktinfo->parse_failed = 1;
		}
	}
	return;
}

static int tuple2stream_htable_key_get_v4_by_packet(struct pkt_info *pktinfo, struct stream_tuple4_v4 *key, int *reversed){
	if(pktinfo->iphdr.v4->saddr < pktinfo->iphdr.v4->daddr){
		key->saddr = pktinfo->iphdr.v4->saddr;
		key->daddr = pktinfo->iphdr.v4->daddr;
		key->source = pktinfo->tcphdr->source;
		key->dest = pktinfo->tcphdr->dest;
		*reversed = 0;
	}
	else{
		key->saddr = pktinfo->iphdr.v4->daddr;
		key->daddr = pktinfo->iphdr.v4->saddr;
		key->source = pktinfo->tcphdr->dest;
		key->dest = pktinfo->tcphdr->source;
		*reversed = 1;
	}
	return 0;
}

static int tuple2stream_htable_key_get_v6_by_packet(struct pkt_info *pktinfo, struct stream_tuple4_v6 *key, int *reversed){
	if(memcmp((void*)&(pktinfo->iphdr.v6->ip6_src), (void*)&(pktinfo->iphdr.v6->ip6_dst), sizeof(key->saddr)) < 0){
		memcpy(key->saddr, &(pktinfo->iphdr.v6->ip6_src), sizeof(key->saddr));
		memcpy(key->daddr, &(pktinfo->iphdr.v6->ip6_dst), sizeof(key->daddr));
		key->source = pktinfo->tcphdr->source;
		key->dest = pktinfo->tcphdr->dest;
		*reversed = 0;
	}
	else{
		memcpy(key->saddr, &(pktinfo->iphdr.v6->ip6_dst), sizeof(key->saddr));
		memcpy(key->daddr, &(pktinfo->iphdr.v6->ip6_src), sizeof(key->daddr));
		key->source = pktinfo->tcphdr->dest;
		key->dest = pktinfo->tcphdr->source;
		*reversed = 1;
	}
	return 0;
}

static int tuple2stream_htable_key_get_v4_by_stream(const struct streaminfo *stream, struct stream_tuple4_v4 *key, int *reversed){
	if(stream->addr.tuple4_v4->saddr < stream->addr.tuple4_v4->daddr){
		key->saddr = stream->addr.tuple4_v4->saddr;
		key->daddr = stream->addr.tuple4_v4->daddr;
		key->source = stream->addr.tuple4_v4->source;
		key->dest = stream->addr.tuple4_v4->dest;
		*reversed = 0;
	}
	else{
		key->saddr = stream->addr.tuple4_v4->daddr;
		key->daddr = stream->addr.tuple4_v4->saddr;
		key->source = stream->addr.tuple4_v4->dest;
		key->dest = stream->addr.tuple4_v4->source;
		*reversed = 1;
	}
	return 0;
}

static int tuple2stream_htable_key_get_v6_by_stream(const struct streaminfo *stream, struct stream_tuple4_v6 *key, int *reversed){
	if(memcmp(stream->addr.tuple4_v6->saddr, stream->addr.tuple4_v6->daddr, sizeof(key->saddr)) < 0){
		memcpy(key->saddr, stream->addr.tuple4_v6->saddr, sizeof(key->saddr));
		memcpy(key->daddr, stream->addr.tuple4_v6->daddr, sizeof(key->daddr));
		key->source = stream->addr.tuple4_v6->source;
		key->dest = stream->addr.tuple4_v6->dest;
		*reversed = 0;
	}
	else{
		memcpy(key->saddr, stream->addr.tuple4_v6->daddr, sizeof(key->saddr));
		memcpy(key->daddr, stream->addr.tuple4_v6->saddr, sizeof(key->daddr));
		key->source = stream->addr.tuple4_v6->dest;
		key->dest = stream->addr.tuple4_v6->source;
		*reversed = 1;
	}
	return 0;
}


static int tuple2stream_htable_add(addr_type_t addr_type, struct pkt_info *pktinfo, 
								struct streaminfo *stream, struct pme_info *pmeinfo, int thread_seq){
	MESA_htable_handle tuple2stream_htable = g_kni_handle->threads_handle[thread_seq].tuple2stream_htable;
	void *logger = g_kni_handle->local_logger;
	int ret;
	char key_str[KNI_ADDR_MAX];
	struct tuple2stream_htable_value *value = ALLOC(struct tuple2stream_htable_value, 1);
	value->stream = stream;
	value->pmeinfo = pmeinfo;
	value->route_dir = stream->routedir;
	//ipv6
	if(addr_type == ADDR_TYPE_IPV6){
		struct stream_tuple4_v6 key;
		tuple2stream_htable_key_get_v6_by_packet(pktinfo, &key, &(value->reversed));
		ret = MESA_htable_add(tuple2stream_htable, (const unsigned char *)&key, sizeof(key), (const void*)value);
		if(ret < 0){		
			kni_addr_trans_v6(&key, key_str, sizeof(key_str));
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_TUPLE2STM_ADD_FAIL], 0, FS_OP_ADD, 1);
			KNI_LOG_ERROR(logger, "MESA_htable: Failed at add, table = tuple2stream_htable, key = %s, key_size = %d, ret = %d",
				key_str, sizeof(key), ret);
		}
	}
	//ipv4
	else{
		struct stream_tuple4_v4 key;
		tuple2stream_htable_key_get_v4_by_packet(pktinfo, &key, &(value->reversed));
		ret = MESA_htable_add(tuple2stream_htable, (const unsigned char *)&key, sizeof(key), (const void*)value);
		if(ret < 0){
			kni_addr_trans_v4(&key, key_str, sizeof(key_str));
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_TUPLE2STM_ADD_FAIL], 0, FS_OP_ADD, 1);
			KNI_LOG_ERROR(logger, "MESA_htable: Failed at add, table = tuple2stream_htable, key = %s, key_size = %d, ret = %d",
				key_str, sizeof(key), ret);
		}
	}
	if(ret >= 0){
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_TUPLE2STM_ADD_SUCC], 0, FS_OP_ADD, 1);
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->line_ids[1], g_kni_fs_handle->column_ids[thread_seq], FS_OP_ADD, 1);
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->line_ids[1], g_kni_fs_handle->column_ids[g_kni_handle->thread_count], FS_OP_ADD, 1);
	}
	return ret;
}

static int traceid2pme_htable_add(struct pme_info *pmeinfo){
	void *logger = g_kni_handle->local_logger;
	int key_size = 0, ret;
	key_size = strnlen(pmeinfo->stream_traceid, sizeof(pmeinfo->stream_traceid));
	ret = MESA_htable_add(g_kni_handle->traceid2pme_htable, (const unsigned char *)(pmeinfo->stream_traceid), 
		key_size, (const void*)pmeinfo);
	if(ret < 0){
		KNI_LOG_ERROR(logger, "MESA_htable: Failed at add,"
				"table = traceid2pme_htable, key = %s, ret = %d", pmeinfo->stream_traceid, ret);
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_ID2PME_ADD_FAIL], 0, FS_OP_ADD, 1);
	}
	else{
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_ID2PME_ADD_SUCC], 0, FS_OP_ADD, 1);
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_ID2PME_CNT], 0, FS_OP_ADD, 1);
	}
	return ret;
}

int tuple2stream_htable_del(const struct streaminfo *stream, int thread_seq){
	MESA_htable_handle handle = g_kni_handle->threads_handle[thread_seq].tuple2stream_htable;
	void *logger = g_kni_handle->local_logger;
	int reversed = 0, ret = -1;
	char key_str[KNI_ADDR_MAX];
	//ipv6
	if(stream->addr.addrtype == ADDR_TYPE_IPV6){
		struct stream_tuple4_v6 key;
		tuple2stream_htable_key_get_v6_by_stream(stream, &key, &reversed);
		ret = MESA_htable_del(handle, (const unsigned char *)(&key), 
						sizeof(key), NULL);
		if(ret < 0){
			kni_addr_trans_v6(&key, key_str, sizeof(key_str));
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_TUPLE2STM_DEL_FAIL], 0, FS_OP_ADD, 1);
			KNI_LOG_ERROR(logger, "MESA_htable: Failed at del, table = %s, key = %s, key_size = %d, ret = %d",
							"tuple2stream_htable", key_str, sizeof(key), ret);
		}
	}
	//ipv4
	else{
		struct stream_tuple4_v4 key;
		tuple2stream_htable_key_get_v4_by_stream(stream, &key, &reversed);
		ret = MESA_htable_del(handle, (const unsigned char *)(&key), sizeof(key), NULL);
		if(ret < 0){
			kni_addr_trans_v4(&key, key_str, sizeof(key_str));
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_TUPLE2STM_DEL_FAIL], 0, FS_OP_ADD, 1);
			KNI_LOG_ERROR(logger, "MESA_htable: Failed at del, table = %s, key = %s, key_size = %d, ret = %d",
							"tuple2stream_htable", key_str, sizeof(key), ret);
		}
	}
	if(ret >= 0){
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_TUPLE2STM_DEL_SUCC], 0, FS_OP_ADD, 1);
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->line_ids[1], g_kni_fs_handle->column_ids[thread_seq], FS_OP_ADD, -1);
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->line_ids[1], g_kni_fs_handle->column_ids[g_kni_handle->thread_count], FS_OP_ADD, -1);
	}
	return ret;
}

static void traceid2pme_htable_del(struct pme_info *pmeinfo){
	//del traceid2pme htable
	if(pmeinfo->action == KNI_ACTION_INTERCEPT){
		void *logger = g_kni_handle->local_logger;
		int key_size = strnlen(pmeinfo->stream_traceid, sizeof(pmeinfo->stream_traceid));
		int ret;
		ret = MESA_htable_del(g_kni_handle->traceid2pme_htable, (const unsigned char *)pmeinfo->stream_traceid, 
						key_size, NULL);
		if(ret < 0){
			KNI_LOG_ERROR(logger, "MESA_htable: Failed at del, table = %s, key = %s, key_size = %d, ret = %d",
							"traceid2pme_htable", pmeinfo->stream_traceid, key_size, ret);
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_ID2PME_DEL_FAIL], 0, FS_OP_ADD, 1);
		}
		else{
			//KNI_LOG_DEBUG(logger, "MESA_htable: Succeed at del, table = %s, key = %s, key_size = %d",
			//					"traceid2pme_htable", pmeinfo->stream_traceid, key_size);
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_ID2PME_DEL_SUCC], 0, FS_OP_ADD, 1);
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_ID2PME_CNT], 0, FS_OP_ADD, -1);
		}
	}
}

static int dabloom_add(struct pkt_info *pktinfo, int thread_seq){
	void *logger = g_kni_handle->local_logger;
	struct dup_traffic_dabloom_key bloom_key;
	memset(&bloom_key, 0, sizeof(bloom_key));
	dup_traffic_dabloom_key_get(pktinfo, &bloom_key);
	int ret = expiry_dablooms_add(g_kni_handle->threads_handle[thread_seq].dabloom_handle, (const char*)&bloom_key, sizeof(bloom_key));
	if(ret < 0){
		KNI_LOG_ERROR(logger, "Failed at expiry_dablooms_add, errmsg = %s", expiry_dablooms_errno_trans((enum expiry_dablooms_errno)ret));
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_BLOOM_ADD_FAIL], 0, FS_OP_ADD, 1);
	}
	//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_BLOOM_ADD_SUCC], 0, FS_OP_ADD, 1);
	uint64_t count = 0;
	expiry_dablooms_element_count_get(g_kni_handle->threads_handle[thread_seq].dabloom_handle, &count);
	FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->line_ids[0], g_kni_fs_handle->column_ids[thread_seq], FS_OP_SET, count);
	return ret;
}

static struct session_runtime_attribute * kni_pull_session_attribute_results(struct streaminfo *a_stream, struct pme_info *pmeinfo)
{
	const struct session_runtime_attribute *session_attr = NULL;
	void *logger = g_kni_handle->local_logger;
	session_attr = session_runtime_attribute_get(a_stream);
	if(session_attr == NULL)
	{
		KNI_LOG_ERROR(logger, "share-session-attribute: Failed to get the session attribute results,stream traceid = %s", pmeinfo->stream_traceid);
		return NULL;
	}
	KNI_LOG_DEBUG(logger, "share-session-attribute: Success to get the session attribute results,stream traceid = %s", pmeinfo->stream_traceid);

	const char *client_subscribe_id = srt_attribute_get_client_subscriber_id(session_attr);
	if(client_subscribe_id == NULL)
		KNI_LOG_DEBUG(logger, "share-session-attribute: source subscribe id is NULL,stream traceid = %s", pmeinfo->stream_traceid);
	else
		KNI_LOG_DEBUG(logger, "share-session-attribute: source subscribe id is: %s,stream traceid = %s", client_subscribe_id, pmeinfo->stream_traceid);

	const char *server_subscribe_id = srt_attribute_get_server_subscriber_id(session_attr);
	if(server_subscribe_id == NULL)
		KNI_LOG_DEBUG(logger, "share-session-attribute: destination subscribe id is NULL,stream traceid = %s", pmeinfo->stream_traceid);
	else
		KNI_LOG_DEBUG(logger, "share-session-attribute: destination subscribe id is: %s,stream traceid = %s", server_subscribe_id, pmeinfo->stream_traceid);

	const struct asn_info *client_asn = srt_attribute_get_client_ip_asn(session_attr);
	if(client_asn == NULL)
	{
		KNI_LOG_DEBUG(logger, "share-session-attribute: source asn and organization is NULL,stream traceid = %s", pmeinfo->stream_traceid);
	}
	else
	{
		if(client_asn->asn_id == NULL)
			KNI_LOG_DEBUG(logger, "share-session-attribute: source asn is NULL,stream traceid = %s", pmeinfo->stream_traceid);
		else
			KNI_LOG_DEBUG(logger, "share-session-attribute: source asn is: %s,stream traceid = %s", client_asn->asn_id, pmeinfo->stream_traceid);	

		if(client_asn->organization == NULL)
			KNI_LOG_DEBUG(logger, "share-session-attribute: source organization is: NULL,stream traceid = %s", pmeinfo->stream_traceid);
		else
			KNI_LOG_DEBUG(logger, "share-session-attribute: source organization is: %s,stream traceid = %s", client_asn->organization, pmeinfo->stream_traceid);
	}

	const struct asn_info *server_asn = srt_attribute_get_server_ip_asn(session_attr);
	if(server_asn == NULL)
	{
		KNI_LOG_DEBUG(logger, "share-session-attribute: destination asn and organization is NULL,stream traceid = %s", pmeinfo->stream_traceid);
	}
	else
	{
		if(server_asn->asn_id == NULL)
			KNI_LOG_DEBUG(logger, "share-session-attribute: destination asn: NULL,stream traceid = %s",pmeinfo->stream_traceid);
		else
			KNI_LOG_DEBUG(logger, "share-session-attribute: destination asn:%s,stream traceid = %s", server_asn->asn_id, pmeinfo->stream_traceid);

		if(server_asn->organization == NULL)
			KNI_LOG_DEBUG(logger, "share-session-attribute: destination organization NULL,stream traceid = %s",pmeinfo->stream_traceid);
		else
			KNI_LOG_DEBUG(logger, "share-session-attribute: destination organization:%s,stream traceid = %s", server_asn->organization, pmeinfo->stream_traceid);
	}

	const struct location_info *client_location = srt_attribute_get_client_ip_location(session_attr);
	if(client_location == NULL)
	{
		KNI_LOG_DEBUG(logger, "share-session-attribute: source ip location is NULL,stream traceid = %s",pmeinfo->stream_traceid);
	}
	else
	{
		if(client_location->full_location == NULL)
			KNI_LOG_DEBUG(logger, "share-session-attribute: source ip full location is NULL,stream traceid = %s",pmeinfo->stream_traceid);
		else
			KNI_LOG_DEBUG(logger, "share-session-attribute: source ip full location is: %s, stream traceid = %s",client_location->full_location, pmeinfo->stream_traceid);
	}

	const struct location_info *server_location = srt_attribute_get_server_ip_location(session_attr);
	if(server_location == NULL)
	{
		KNI_LOG_DEBUG(logger, "share-session-attribute: destination ip location is NULL,stream traceid = %s",pmeinfo->stream_traceid);
	}
	else
	{
		if(server_location->full_location == NULL)
			KNI_LOG_DEBUG(logger, "share-session-attribute: destination ip full location is NULL,stream traceid = %s",pmeinfo->stream_traceid);
		else
			KNI_LOG_DEBUG(logger, "share-session-attribute: destination ip full location is: %s, stream traceid = %s",server_location->full_location, pmeinfo->stream_traceid);
	}

	const char *ja3_fingerprint = srt_attribute_get_ja3_fingerprint(session_attr);
	if(ja3_fingerprint == NULL)
		KNI_LOG_DEBUG(logger, "share-session-attribute: ja3_fingerprint is NULL,stream traceid = %s",pmeinfo->stream_traceid);
	else
		KNI_LOG_DEBUG(logger, "share-session-attribute: ja3_fingerprint is %s,stream traceid = %s", ja3_fingerprint, pmeinfo->stream_traceid);

	unsigned int category_ids[8] = {0};

	int n_category_ids = srt_attribute_get_category_ids(session_attr, category_ids, sizeof(category_ids)/sizeof(category_ids[0]));

	KNI_LOG_DEBUG(logger, "share-session-attribute: fqdn_category_id number is %d,stream traceid = %s", n_category_ids, pmeinfo->stream_traceid);
	if(n_category_ids < 0 || n_category_ids > 8)
	{
		KNI_LOG_DEBUG(logger, "share-session-attribute: fqdn_category_id_num out of range( 0 <= value <= 8), value = %d,stream traceid = %s", n_category_ids, pmeinfo->stream_traceid);
	}
	else
	{
		for(int i= 0; i < n_category_ids; i ++)
		{
			KNI_LOG_DEBUG(logger, "share-session-attribute: fqdn_category_id[%d] = %u,stream traceid = %s",i, category_ids[i],pmeinfo->stream_traceid);
		}
	}

	return (struct session_runtime_attribute *)session_attr;
}


static int tsg_diagnose_judge_streamshunt(int maat_rule_config_id,struct pme_info *pmeinfo)
{
	int i = 0 ,ret = 0;
	void *logger = g_kni_handle->local_logger;

	if(g_kni_handle->tsg_diagnose_enable == 0){
		KNI_LOG_DEBUG(logger, "Tsg diagnose: enabled is 0, stream traceid = %s, stream addr = %s", pmeinfo->stream_traceid, pmeinfo->stream_addr);
		return 0;
	}


	if(g_kni_handle->secpolicyid_shunt_tsg_diagnose.id_num == 0){
		KNI_LOG_DEBUG(logger, "Tsg diagnose: no security policy from profile to shunt, stream traceid = %s, stream addr = %s", pmeinfo->stream_traceid, pmeinfo->stream_addr);
		return 0;
	}
	for(i = 0; i < g_kni_handle->secpolicyid_shunt_tsg_diagnose.id_num; i ++){
		if(g_kni_handle->secpolicyid_shunt_tsg_diagnose.id_arr[i] == 0){
			KNI_LOG_DEBUG(logger, "Tsg diagnose: security policy 0 is not allowd shunt, stream traceid = %s, stream addr = %s", pmeinfo->stream_traceid, pmeinfo->stream_addr);
			continue;
		}
		if(g_kni_handle->secpolicyid_shunt_tsg_diagnose.id_arr[i] == maat_rule_config_id){
			ret = 1;
			KNI_LOG_DEBUG(logger, "Tsg diagnose: security policy id %d shunt, stream traceid = %s, stream addr = %s", maat_rule_config_id, pmeinfo->stream_traceid, pmeinfo->stream_addr);
			break;
		}
	}
	return ret;
}
static void set_timestamp_depend_first_data(struct streaminfo *stream, struct pme_info *pmeinfo, struct pkt_info *pktinfo){
	struct kni_tcpopt_info tcpopt_first_data;
	if(pmeinfo->client_tcpopt.ts_set != 1 || pmeinfo->server_tcpopt.ts_set != 1)
		return;
	kni_get_tcpopt(&tcpopt_first_data,pktinfo->tcphdr,pktinfo->tcphdr_len);
	if(stream->curdir == 1){
		pmeinfo->client_tcpopt.ts_value = tcpopt_first_data.ts_value;
		pmeinfo->server_tcpopt.ts_value = tcpopt_first_data.ts_ecr;
	}
	if(stream->curdir == 2){
		pmeinfo->client_tcpopt.ts_value = tcpopt_first_data.ts_ecr;
		pmeinfo->server_tcpopt.ts_value = tcpopt_first_data.ts_value;
	}
}

static int is_intercept_error(struct streaminfo *stream, struct pme_info *pmeinfo,
							  struct pkt_info *pktinfo,  int thread_seq)
{
    void *logger = g_kni_handle->local_logger;
    int ret;
    unsigned char intercept_stream_link_mode;
    int intercept_stream_link_mode_len = sizeof(unsigned char);
    unsigned short stream_tunnel_type = STREAM_TUNNLE_NON;
    int stream_tunnel_type_len = sizeof(unsigned short);
    int has_dup_traffic;
    int have_dup_pkt_len = sizeof(has_dup_traffic);

    //intercept_error: TCP CTEAT LINK NOT BYSYN or TCP_CREATE_LINK_MODE error
    ret = MESA_get_stream_opt(stream, MSO_TCP_CREATE_LINK_MODE, (void *)&intercept_stream_link_mode, 
                              &intercept_stream_link_mode_len);
    if(ret == 0)
    {
        if(intercept_stream_link_mode != TCP_CTEAT_LINK_BYSYN)
        {
            KNI_LOG_DEBUG(logger, "Intercept error: TCP_CREATE_LINK_MODE is not BYSYN, link_mode=%d, link_mode_len=%d, "
                                   "stream traceid = %s, stream addr = %s", 
                         intercept_stream_link_mode, intercept_stream_link_mode_len, pmeinfo->stream_traceid, 
                         pmeinfo->stream_addr);
            pmeinfo->intcp_error = INTERCEPT_ERROR_NOT_TCP_LINK_BYSYN;
            FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_NOT_LINK_MODE_BYSYN], 0, 
                       FS_OP_ADD, 1);
            
            return 1;
        }
    }
    else
    {
        KNI_LOG_DEBUG(logger, "Intercept error: get MSO_TCP_CREATE_LINK_MODE error, ret = %d, stream traceid = %s, "
                              "stream addr = %s",
                      ret, pmeinfo->stream_traceid, pmeinfo->stream_addr);
        pmeinfo->intcp_error = INTERCEPT_ERROR_GET_TCP_LINK_MODE_ERR;
        FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_GET_LINK_MODE_ERR], 0, 
                   FS_OP_ADD, 1);
        
        return 1;
    }

    //intercept_error: TCP_TUNNEL_TYPE error
    ret = MESA_get_stream_opt(stream, MSO_STREAM_TUNNEL_TYPE, (void *)&stream_tunnel_type, &stream_tunnel_type_len);
    if(ret == 0)
    {
        if(stream_tunnel_type != STREAM_TUNNLE_NON)
        {
            KNI_LOG_DEBUG(logger, "Intercept error: stream type is tunnel, STREAM_TUNNLE_TYPE = %d, "
                                  "stream traceid = %s, stream addr = %s",
                          stream_tunnel_type, pmeinfo->stream_traceid, pmeinfo->stream_addr);
            pmeinfo->intcp_error = INTERCEPT_ERROR_STREAM_TUNNLE_TYPE;
            FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_STREAM_IS_TUN_TYPE], 0, 
            FS_OP_ADD, 1);

            return 1;
        }
    }
    else{
        KNI_LOG_DEBUG(logger, "Intercept error: get MSO_STREAM_TUNNEL_TYPE error, ret = %d, stream traceid = %s, "
                              "stream addr = %s",
                      ret, pmeinfo->stream_traceid, pmeinfo->stream_addr);
        pmeinfo->intcp_error = INTERCEPT_ERROR_GET_STREAM_TUNNLE_TYPE_ERR;
        FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_GET_STREAM_TUN_TYPE_ERR], 0, 
                   FS_OP_ADD, 1);

        return 1;
    }

    //intercept_error: not double dir
    if(stream->dir != DIR_DOUBLE)
    {
        KNI_LOG_DEBUG(logger, "Intercept error: asym routing, stream traceid = %s, stream addr = %s", 
                      pmeinfo->stream_traceid, pmeinfo->stream_addr);
        FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_ASYM_ROUTING], 0, FS_OP_ADD, 1);
        pmeinfo->intcp_error = INTERCEPT_ERROR_ASYM_ROUTING;
        
        return 1;
    }

    //intercept_error: no syn
    if(pmeinfo->has_syn == 0)
    {
        KNI_LOG_DEBUG(logger, "Intercept error: no syn, stream traceid = %s, stream addr = %s", pmeinfo->stream_traceid,
                      pmeinfo->stream_addr);
        FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_NO_SYN], 0, FS_OP_ADD, 1);
        pmeinfo->intcp_error = INTERCEPT_ERROR_NO_SYN;

        return 1;
    }

    //intercept_error: no syn/ack
    if(pmeinfo->has_syn_ack == 0)
    {
        KNI_LOG_DEBUG(logger, "Intercept error: no syn/ack, stream traceid = %s, stream addr = %s", 
                      pmeinfo->stream_traceid, pmeinfo->stream_addr);
        FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_NO_SYN_ACK], 0, FS_OP_ADD, 1);
        pmeinfo->intcp_error = INTERCEPT_ERROR_NO_SYN_ACK;

        return 1;
    }

    //intercept_error: invalid ip header
    if(pktinfo->parse_failed == 1)
    {
        pmeinfo->intcp_error = INTERCEPT_ERROR_INVALID_IP_HDR;
        KNI_LOG_DEBUG(logger, "Intercept error: invalid ip header, stream traceid = %s, stream addr = %s", 
                                    pmeinfo->stream_traceid, pmeinfo->stream_addr);
        FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_INVALID_IP_HDR],0, FS_OP_ADD, 1);

        return 1;
    }

    //intercept_error: first data > 1500, bypass and dropme 
    if(pktinfo->ip_totlen > KNI_DEFAULT_MTU)
    {
        pmeinfo->intcp_error = INTERCEPT_ERROR_EXCEED_MTU;
        KNI_LOG_DEBUG(logger, "Intercept error: first data packet exceed MTU(1500), stream traceid = %s, "
                               "stream addr = %s", 
                      pmeinfo->stream_traceid, pmeinfo->stream_addr);
        FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_EXCEED_MTU], 0, FS_OP_ADD, 1);

        return 1;
    }

    //intercept_error: no tfe
    if(tsg_diagnose_judge_streamshunt(pmeinfo->maat_rule.rule_id, pmeinfo) == 0) // tsg  diagnose shunt
    {
        pmeinfo->tfe_id = tfe_mgr_alive_node_get(g_kni_handle->_tfe_mgr, thread_seq);
    }
    else
    {
        pmeinfo->tfe_id = tfe_mgr_alive_node_cycle_get(g_kni_handle->_tfe_mgr,
                                                       (int *)&(g_kni_handle->arr_last_tfe_dispatch_index[thread_seq]));
    }
    if(pmeinfo->tfe_id < 0)
    {
        KNI_LOG_DEBUG(logger, "Intercept error: no available tfe, stream traceid = %s, stream addr = %s", 
                      pmeinfo->stream_traceid, pmeinfo->stream_addr);
        pmeinfo->intcp_error = INTERCEPT_ERROR_NO_TFE;
        FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_NO_TFE], 0, FS_OP_ADD, 1);
        return 1;
    }

    //intercept_error: dup traffic
	if(g_kni_handle->dup_traffic_switch == 1)
    {
		//has dup traffic
		if(pmeinfo->has_dup_syn == 1 || pmeinfo->has_dup_syn_ack == 1)
        {
			pmeinfo->has_dup_traffic = 1;
		}
		if(pmeinfo->has_dup_traffic == 1)
        {
			if(g_kni_handle->dup_traffic_action == KNI_ACTION_BYPASS)
            {
				KNI_LOG_DEBUG(g_kni_handle->local_logger, "Intercept error: stream has dup traffic, dup_traffic_action = bypass, "
                                                           "stream traceid = %s, stream addr = %s", 
                              pmeinfo->stream_traceid, pmeinfo->stream_addr);
				FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_DUP_TRAFFIC], 0, 
                           FS_OP_ADD, 1);
				pmeinfo->intcp_error = INTERCEPT_ERROR_DUP_TRAFFIC;

				return 1;
			}
		}
	}

    // get HAVE_DUP_PKT field
	ret = MESA_get_stream_opt(stream, MSO_HAVE_DUP_PKT, (void *)&has_dup_traffic, &have_dup_pkt_len);
	if(ret != 0)
    {
		KNI_LOG_DEBUG(logger, "Intercept error: get MSO_HAVE_DUP_PKT error, ret = %d, stream traceid = %s, "
                              "stream addr = %s",
                      ret, pmeinfo->stream_traceid, pmeinfo->stream_addr);
		pmeinfo->intcp_error = INTERCEPT_ERROR_GET_HAVE_DUP_PKT_ERR;
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_GET_HAVE_DUP_PKT_ERR], 0,
                   FS_OP_ADD, 1);
		
        return 1;
	}
	else
    {
		if(has_dup_traffic == -2){
			KNI_LOG_ERROR(logger, "Intercept error: has duplicate traffic is not sure,has_dup_traffic = %d,"
                                  "stream traceid = %s, stream addr = %s",
                          has_dup_traffic, pmeinfo->stream_traceid, pmeinfo->stream_addr);
			pmeinfo->intcp_error = INTERCEPT_ERROR_DUP_PKT_NOT_SURE_ERR;
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_DUP_PKT_NOT_SURE_ERR], 0,
                       FS_OP_ADD, 1);
			
            return 1;
		}
		pmeinfo->has_dup_traffic = (uint64_t)has_dup_traffic;
	}


	//Bypass Duplicated Packet
	if(g_kni_handle->pxy_tcp_option_enable == 1)
	{
		if(pmeinfo->has_dup_traffic == 1 && pmeinfo->pxy_tcp_option.bypass_duplicated_packet == 1)
		{
			KNI_LOG_DEBUG(g_kni_handle->local_logger, "Proxy-tcp-option: bypass Duplicated Packet first data, "
                                                      "streamid = %d", 
                          pmeinfo->stream_traceid);
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_DUP_TRAFFIC], 0, 
                       FS_OP_ADD, 1);

			return 1;
		}
	}

    return 0;
}


static int is_dynamic_bypass(struct streaminfo *stream, struct pme_info *pmeinfo, struct pkt_info *pktinfo, 
                             int thread_seq)
{
    int ret = 0;
    void *logger = g_kni_handle->local_logger;

	if(g_kni_handle->ssl_dynamic_bypass_enable == 1)
    {
		if(first_data_ssl_dynamic_bypass(stream, pmeinfo, pktinfo, thread_seq) == 0)
		{
			//dynamic bypass fs stat
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_DY_PASS_STM], 0, 
                       FS_OP_ADD, 1);
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_DY_PASS_BYTE], 0,
                       FS_OP_ADD, pktinfo->ip_totlen);
			//dynamic bypass ipv4 or ipv6
			if(stream->addr.addrtype == ADDR_TYPE_IPV6)
            {
				FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_DY_PASS_IPV6_STM], 0, 
                           FS_OP_ADD, 1);
			}
			else
            {
				FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_DY_PASS_IPV4_STM], 0,
                           FS_OP_ADD, 1);
			}

			if(pmeinfo->has_dup_traffic == 1)
            {
				FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_DUP_TFC_STM], 0,
                           FS_OP_ADD, 1);
				KNI_LOG_DEBUG(logger, "stream has dup traffic, traceid = %s", pmeinfo->stream_traceid);
			}

			pmeinfo->ssl_intercept_state = 0;
			pmeinfo->is_dynamic_bypass = 1;

			if(g_kni_handle->dup_traffic_switch == 1 && pmeinfo->has_dup_traffic == 1)
            {
                ret = dabloom_add(pktinfo, thread_seq);
                if(ret < 0)
                {
                    KNI_LOG_DEBUG(logger, "stream add dabloom fail, ret=%d, traceid = %s",ret, 
                                  pmeinfo->stream_traceid);
                }
			}
			//return APP_STATE_FAWPKT | APP_STATE_GIVEME;
            return 1;
		}
	}

    return 0;
}


static void set_proxy_metric_value_by_packet(const struct streaminfo *stream, 
                                             struct proxy_metric_value *value)
{
    int ret = 0, i_or_e = 0, pkt_len = 0;

    ret = get_rawpkt_opt_from_streaminfo(stream, RAW_PKT_GET_TOT_LEN, &pkt_len);
    if(ret < 0)
    {
        return;
    }

    i_or_e = MESA_dir_link_to_human(stream->routedir);
    switch(i_or_e)
    {
        case 'E':
        case 'e':
            value->in_bytes += pkt_len;
            value->in_pkts++;
            break;
        case 'I':
        case 'i':
            value->out_bytes += pkt_len;
            value->out_pkts++;
            break;
        default:
			return;
    }

    if(value->in_pkts != 0 || value->out_pkts != 0)
    {
        value->hit_count = 1;
    }

	return;
}

static int is_send_to_tfe_failed(struct streaminfo *stream, struct pme_info *pmeinfo,
                                 struct pkt_info *pktinfo, int thread_seq)
{
    int len = 0, ret = 0;
    char *buff = NULL;
    void *logger = g_kni_handle->local_logger;

	//session attribute NULL
	if(pmeinfo->session_attribute == NULL)
	{
		KNI_LOG_DEBUG(logger, "Intercept error: Get share session attribute error,"
                              "stream traceid = %s", 
                      pmeinfo->stream_traceid);
        FS_operate(g_kni_fs_handle->handle, 
                   g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_CMSG_ADD_FAIL],
                   0, FS_OP_ADD, 1);
        pmeinfo->intcp_error = INTERCEPT_ERROR_CMSG_ADD_FAIL;

        goto error;
	}

    //get intercept success first data timestamps send to tfe 
    set_timestamp_depend_first_data(stream, pmeinfo, pktinfo);

    //add cmsg info to packet
    buff = add_cmsg_to_packet(pmeinfo, stream, pktinfo, &len);
    if(buff == NULL)
    {
        KNI_LOG_DEBUG(logger, "Intercept error: failed at add cmsg to packet, "
                              "stream traceid = %s, stream addr = %s", 
                      pmeinfo->stream_traceid, pmeinfo->stream_addr);
        FS_operate(g_kni_fs_handle->handle, 
                   g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_CMSG_ADD_FAIL],
                   0, FS_OP_ADD, 1);
        pmeinfo->intcp_error = INTERCEPT_ERROR_CMSG_ADD_FAIL;

        goto error;
    }

    ret = tuple2stream_htable_add(pmeinfo->addr_type, pktinfo, stream, pmeinfo, 
                                  thread_seq);
    if(ret < 0)
    {
        KNI_LOG_DEBUG(logger, "Intercept error: tuple2stm add fail, "
                              "stream traceid = %s, stream addr = %s", 
                      pmeinfo->stream_traceid, pmeinfo->stream_addr);
        FS_operate(g_kni_fs_handle->handle, 
                   g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_TUPLE2STM_ADD_FAIL],
                   0, FS_OP_ADD, 1);
        pmeinfo->intcp_error = INTERCEPT_ERROR_TUPLE2STM_ADD_FAIL;

        goto  error;
    }

    //Note: traceid2pme_add_fail, still work. no cmsg
    traceid2pme_htable_add(pmeinfo);

    //send to tfe
    ret = send_to_tfe(buff, len, thread_seq, pmeinfo->tfe_id, pmeinfo->addr_type, logger);
    if(ret < 0)
    {
        KNI_LOG_DEBUG(logger, "Intercept error: failed at send first packet to tfe%d, stream traceid = %s, "
                              "stream addr = %s", 
                      pmeinfo->tfe_id, pmeinfo->stream_traceid, pmeinfo->stream_addr);
        FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_SENDTO_TFE_FAIL], 0, 
                   FS_OP_ADD, 1);
        pmeinfo->intcp_error = INTERCEPT_ERROR_SENDTO_TFE_FAIL;

        tuple2stream_htable_del(stream, thread_seq);
        traceid2pme_htable_del(pmeinfo);

        goto error;
    }
    else{
        KNI_LOG_DEBUG(logger, "Succeed at send first packet to tfe%d, stream traceid = %s, stream addr = %s",
                      pmeinfo->tfe_id, pmeinfo->stream_traceid, pmeinfo->stream_addr);
    }

	if(buff != NULL)
    {
		FREE(&buff);
	}
    return 0;

error:
	if(buff != NULL)
    {
		FREE(&buff);
	}
    return 1;
}


static void set_first_data_intercep_fs(struct streaminfo *stream, struct pme_info *pmeinfo)
{
	void *logger = g_kni_handle->local_logger;

    FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCP_STM], 0, FS_OP_ADD, 1);

	//ipv4 or ipv6
	if(stream->addr.addrtype == ADDR_TYPE_IPV6)
    {
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_IPV6_STM], 0, FS_OP_ADD, 1);
	}
	else
    {
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_IPV4_STM], 0, FS_OP_ADD, 1);
	}
	//http or ssl
	if(pmeinfo->protocol == PROTO_SSL)
    {
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_SSL_STM], 0, FS_OP_ADD, 1);
	}	
	if(pmeinfo->protocol == PROTO_HTTP)
    {
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_HTTP_STM], 0, FS_OP_ADD, 1);
	}

	//dup_traffic_stm
	if(pmeinfo->has_dup_traffic == 1)
    {
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_DUP_TFC_STM], 0, FS_OP_ADD, 1);
		KNI_LOG_DEBUG(logger, "stream has dup traffic, traceid = %s", pmeinfo->stream_traceid);
	}

}

static int first_data_no_intercept(struct streaminfo *stream, struct pme_info *pmeinfo, 
                                   struct pkt_info *pktinfo, int thread_seq)
{
    void *logger = g_kni_handle->local_logger;
    KNI_LOG_DEBUG(logger, "No_intercept Hit: stream traceid = %s, "
                          "stream addr = %s", 
                  pmeinfo->stream_traceid, pmeinfo->stream_addr);

    return APP_STATE_FAWPKT | APP_STATE_GIVEME;
}


static int first_data_intercept(struct streaminfo *stream, struct pme_info *pmeinfo, 
                                struct pkt_info *pktinfo, int thread_seq)
{

    // if(1 == add_intercept_stream_to_tuple2stream(stream, pmeinfo, pktinfo, thread_seq))
    // {
    //     return APP_STATE_FAWPKT | APP_STATE_DROPME;
    // }


    if(1 == is_intercept_error(stream, pmeinfo, pktinfo, thread_seq))
    {
        return APP_STATE_FAWPKT | APP_STATE_GIVEME;
    }

    if(1 == is_dynamic_bypass(stream, pmeinfo, pktinfo, thread_seq))
    {
        return APP_STATE_FAWPKT | APP_STATE_GIVEME;
    }

    if(1 == is_send_to_tfe_failed(stream, pmeinfo, pktinfo, thread_seq))
    {
        return APP_STATE_FAWPKT | APP_STATE_GIVEME;
    }

	set_first_data_intercep_fs(stream, pmeinfo);

	return APP_STATE_DROPPKT | APP_STATE_GIVEME;
}

static int dabloom_search(struct pkt_info *pktinfo, int thread_seq){
	void *logger = g_kni_handle->local_logger;
	struct dup_traffic_dabloom_key bloom_key;
	memset(&bloom_key, 0, sizeof(bloom_key));
	dup_traffic_dabloom_key_get(pktinfo, &bloom_key);
	int ret = expiry_dablooms_search(g_kni_handle->threads_handle[thread_seq].dabloom_handle, (const char*)&bloom_key, sizeof(bloom_key));
	//ret = 1, = dup packet, bypass the packet 
	if(ret < 0){
		KNI_LOG_ERROR(logger, "Failed at expiry_dablooms_search, errmsg = %s", expiry_dablooms_errno_trans((enum expiry_dablooms_errno)ret));
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_BLOOM_SEARCH_FAIL], 0, FS_OP_ADD, 1);
	}
	else{
		//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_BLOOM_SEARCH_SUCC], 0, FS_OP_ADD, 1);
		if(ret == 1){
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_BLOOM_HIT], 0, FS_OP_ADD, 1);
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_DUP_TFC_BYTE], 0, FS_OP_ADD, pktinfo->ip_totlen);
		}
		else{
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_BLOOM_MISS], 0, FS_OP_ADD, 1);
		}
	}
	uint64_t count = 0;
	expiry_dablooms_element_count_get(g_kni_handle->threads_handle[thread_seq].dabloom_handle, &count);
	FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->line_ids[0], g_kni_fs_handle->column_ids[thread_seq], FS_OP_SET, count);
	return ret;
}

/* action
	0x00: none
	0x02: intercept
	0x80: bypass
*/
char* kni_maat_action_trans(enum kni_action action){
	switch(action){
		case 0x00:
			return (char*)"none";
		case 0x02:
			return (char*)"intercept";
		case 0x03:
			return (char*)"no_intercept";
		case 0x80:
			return (char*)"bypass";
		default:
			return (char*)"unknown";
	}
}

char next_data_intercept(struct streaminfo *stream, struct pme_info *pmeinfo, const void *a_packet, struct pkt_info *pktinfo, int thread_seq){
	//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCP_READY_BYTE], 0, FS_OP_ADD, pktinfo->ip_totlen);
	int ret, len;
	void *logger = g_kni_handle->local_logger;
	struct iphdr *ipv4_hdr = NULL;
	struct ip6_hdr* ipv6_hdr = NULL;

    set_proxy_metric_value_by_packet(stream, &(pmeinfo->proxy_metric_value));

	if(pktinfo->parse_failed == 1){
		KNI_LOG_ERROR(logger, "next_data_intercept: invalid ip header, drop pkt and not send to tfe");
		return APP_STATE_DROPPKT | APP_STATE_GIVEME;
	}
    if(pmeinfo->intcp_error != 0)
    {
        return APP_STATE_FAWPKT | APP_STATE_GIVEME;
    }


	//search dabloom
	if(g_kni_handle->dup_traffic_switch == 1){
		if(pmeinfo->has_dup_traffic == 1){
			//ret = 1, = dup packet, bypass the packet 
			if(g_kni_handle->pxy_tcp_option_enable == 1)
			{
				if(pmeinfo->pxy_tcp_option.bypass_duplicated_packet == 1){ //Bypass Duplicated Packet
					return APP_STATE_FAWPKT | APP_STATE_GIVEME;
				}
			}
			ret = dabloom_search(pktinfo, thread_seq);
			if(ret == 1){
				return APP_STATE_FAWPKT | APP_STATE_GIVEME;
			}
		}
	}
	if(pmeinfo->addr_type == ADDR_TYPE_IPV6){
		ipv6_hdr = (struct ip6_hdr*)a_packet;
		len = ntohs(ipv6_hdr->ip6_ctlun.ip6_un1.ip6_un1_plen) + sizeof(struct ip6_hdr);
	}
	else{
		ipv4_hdr = (struct iphdr*)a_packet;
		len = ntohs(ipv4_hdr->tot_len);
	}
	if(pktinfo->ip_totlen > KNI_DEFAULT_MTU){
		KNI_LOG_DEBUG(logger, "Next data packet exceed MTU(1500), stream traceid = %s, stream addr = %s", 
				pmeinfo->stream_traceid, pmeinfo->stream_addr);
		return APP_STATE_DROPPKT | APP_STATE_GIVEME;
	}
	if(g_kni_handle->ssl_dynamic_bypass_enable == 1){
		if(pmeinfo->is_dynamic_bypass){
			next_data_ssl_dynamic_bypass(pktinfo);
			return APP_STATE_FAWPKT | APP_STATE_GIVEME;
		}
	}

	ret = send_to_tfe((char*)a_packet, len, thread_seq, pmeinfo->tfe_id, pmeinfo->addr_type, logger);
	if(ret < 0){
		KNI_LOG_ERROR(logger, "Failed at send continue packet to tfe%d, stream traceid = %s, stream addr = %s", 
						pmeinfo->tfe_id, pmeinfo->stream_traceid, pmeinfo->stream_addr);
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCPERR_SENDTO_TFE_FAIL], 0, FS_OP_ADD, 1);
	}
	//else{
	//	FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_INTCP_BYTE], 0, FS_OP_ADD, pktinfo->ip_totlen);
	//}
	return APP_STATE_DROPPKT | APP_STATE_GIVEME;
}

static void read_policy_hit_status(struct streaminfo *stream, struct pme_info *pmeinfo, struct pkt_info *pktinfo,
                                  int thread_seq)
{
    int maat_hit = 0;
    size_t n_hit_rule = 0;
    void *logger = g_kni_handle->local_logger;

    n_hit_rule = session_matched_rules_copy(stream, TSG_SERVICE_INTERCEPT, &pmeinfo->maat_rule, 1);

    if(n_hit_rule == 0)
    {
        maat_hit = 0;
        pmeinfo->action = KNI_ACTION_NONE;
        KNI_LOG_INFO(logger, "intercept_policy_scan: %s, %s, maat_hit = %d, stream traceid = %s", 
                     pmeinfo->stream_addr, (char*)&(pmeinfo->domain), maat_hit, pmeinfo->stream_traceid);
    }
    else
    {
        maat_hit = 1;
        const struct session_runtime_process_context * session_context = 
            session_runtime_process_context_get((const struct streaminfo *)stream);

        pmeinfo->n_maat_rule = 1;
        pmeinfo->protocol = srt_process_context_get_protocol(session_context);

        const char *domain = srt_process_context_get_domain(session_context);
        if(domain != NULL)
        {
            pmeinfo->domain_len = MIN(strlen(domain), (int)sizeof(pmeinfo->domain) - 1);
            if(pmeinfo->protocol == PROTO_HTTP)
            {
                strncpy(pmeinfo->domain.host, domain, pmeinfo->domain_len);
            }
            if(pmeinfo->protocol == PROTO_SSL)
            {
                strncpy(pmeinfo->domain.sni, domain, pmeinfo->domain_len);
            }
        }
        else
        {
            pmeinfo->domain_len = 0;	
        }

        //pmeinfo->action = KNI_ACTION_INTERCEPT;
        pmeinfo->action = (enum kni_action)(pmeinfo->maat_rule.action);
        pmeinfo->policy_id = pmeinfo->maat_rule.rule_id;
        pmeinfo->do_log = pmeinfo->maat_rule.do_log;
        pmeinfo->thread_seq = thread_seq;
        pmeinfo->is_dynamic_bypass = 0;
        pmeinfo->session_attribute = kni_pull_session_attribute_results(stream,pmeinfo);

        char *action_str = kni_maat_action_trans(pmeinfo->action);
        KNI_LOG_INFO(logger, "intercept_policy_scan: %s, %s, maat_hit = %d, rule_id = %u, action = %d(%s), "
                             "stream traceid = %s", 
                     pmeinfo->stream_addr, (char*)&(pmeinfo->domain), maat_hit, pmeinfo->policy_id, 
                     pmeinfo->action, action_str, pmeinfo->stream_traceid);
    }
}


static int first_data_hit_policy(struct streaminfo *stream, struct pme_info *pmeinfo, 
                                 struct pkt_info *pktinfo, int thread_seq)
{

    void *logger = g_kni_handle->local_logger;
    int ret = APP_STATE_FAWPKT | APP_STATE_DROPME;

    set_proxy_metric_value_by_packet(stream, &(pmeinfo->proxy_metric_value));


    switch(pmeinfo->action)
    {
        case KNI_ACTION_INTERCEPT:
            pmeinfo->ssl_intercept_state = 1;
            pmeinfo->tld_handle = TLD_create(-1);
            if(g_kni_handle->pxy_tcp_option_enable == 1)
            {
                pxy_tcp_option_param_get(pmeinfo, logger);
            }
            FS_operate(g_kni_fs_handle->handle, 
                       g_kni_fs_handle->fields[KNI_FIELD_INTCP_READY_STM],
                       0, FS_OP_ADD, 1);

            ret = first_data_intercept(stream, pmeinfo, pktinfo, thread_seq);
            break;

        case KNI_ACTION_NO_INTERCEPT:
            pmeinfo->tld_handle = TLD_create(-1);
            FS_operate(g_kni_fs_handle->handle, 
                       g_kni_fs_handle->fields[KNI_FIELD_NO_INTCP_STM],
                       0, FS_OP_ADD, 1);

            ret = first_data_no_intercept(stream, pmeinfo, pktinfo, thread_seq);
            break;

        default:
            assert(0);
            break;
    }
    return ret;
}

char first_data_process(struct streaminfo *stream, struct pme_info *pmeinfo, 
                        struct pkt_info *pktinfo, int thread_seq)
{
    read_policy_hit_status(stream, pmeinfo, pktinfo, thread_seq);

    if(pmeinfo->action == KNI_ACTION_INTERCEPT || 
       pmeinfo->action == KNI_ACTION_NO_INTERCEPT)
    {
        return first_data_hit_policy(stream, pmeinfo, pktinfo, thread_seq);
    }
    else
    {
        return APP_STATE_FAWPKT | APP_STATE_DROPME;
    }
}

void dup_traffic_detect(struct pme_info *pmeinfo, struct pkt_info *pktinfo){
	if(g_kni_handle->dup_traffic_switch == 0){
		return;
	}
	//syn
	if(pktinfo->tcphdr->syn && !pktinfo->tcphdr->ack){
		if(pmeinfo->syn_packet == NULL){
			struct dup_traffic_dabloom_key *syn_packet = ALLOC(struct dup_traffic_dabloom_key, 1);
			dup_traffic_dabloom_key_get(pktinfo, syn_packet);
			pmeinfo->syn_packet = syn_packet;
		}
		else{
			struct dup_traffic_dabloom_key *syn_packet = ALLOC(struct dup_traffic_dabloom_key, 1);
			dup_traffic_dabloom_key_get(pktinfo, syn_packet);
			if(memcmp(pmeinfo->syn_packet, syn_packet, sizeof(*syn_packet)) == 0){
				pmeinfo->has_dup_syn = 1;
			}
			FREE(&(pmeinfo->syn_packet));
			pmeinfo->syn_packet = syn_packet;
		}
	}
	//syn/ack
	if(pktinfo->tcphdr->syn && pktinfo->tcphdr->ack){
		if(pmeinfo->syn_ack_packet == NULL){
			struct dup_traffic_dabloom_key *syn_ack_packet = ALLOC(struct dup_traffic_dabloom_key, 1);
			dup_traffic_dabloom_key_get(pktinfo, syn_ack_packet);
			pmeinfo->syn_ack_packet = syn_ack_packet;
		}
		else{
			struct dup_traffic_dabloom_key *syn_ack_packet = ALLOC(struct dup_traffic_dabloom_key, 1);
			dup_traffic_dabloom_key_get(pktinfo, syn_ack_packet);
			if(memcmp(pmeinfo->syn_ack_packet, syn_ack_packet, sizeof(*syn_ack_packet)) == 0){
				pmeinfo->has_dup_syn_ack = 1;
			}
			FREE(&(pmeinfo->syn_ack_packet));
			pmeinfo->syn_ack_packet = syn_ack_packet;
		}
	}
}

void tcp_handshake_pkt_process(struct pme_info *pmeinfo, struct pkt_info *pktinfo){
	//syn
	if(pktinfo->tcphdr->syn && !pktinfo->tcphdr->ack){
		pmeinfo->client_window = ntohs(pktinfo->tcphdr->window);
		pmeinfo->has_syn = 1;
		kni_get_tcpopt(&(pmeinfo->client_tcpopt), pktinfo->tcphdr, pktinfo->tcphdr_len);
	}
	//syn/ack
	if(pktinfo->tcphdr->syn && pktinfo->tcphdr->ack){
		pmeinfo->server_window = ntohs(pktinfo->tcphdr->window);
		pmeinfo->has_syn_ack = 1;
		kni_get_tcpopt(&(pmeinfo->server_tcpopt), pktinfo->tcphdr, pktinfo->tcphdr_len);
	}	
	dup_traffic_detect(pmeinfo, pktinfo);
}

static char data_opstate(struct streaminfo *stream, struct pme_info *pmeinfo, const void *a_packet, int thread_seq){
	void *logger = g_kni_handle->local_logger;
	//parse ipv4/6 header
	struct pkt_info pktinfo;
	memset(&pktinfo, 0, sizeof(pktinfo));
	wrapped_kni_header_parse(a_packet, pmeinfo, &pktinfo);
	//pmeinfo->action has only 2 value: KNI_ACTION_NONE, KNI_ACTION_INTERCEPT
	if(pmeinfo->action == KNI_ACTION_INTERCEPT){
		return next_data_intercept(stream, pmeinfo, a_packet, &pktinfo, thread_seq);
	}
    if(pmeinfo->action == KNI_ACTION_NO_INTERCEPT)
    {
        set_proxy_metric_value_by_packet(stream, &(pmeinfo->proxy_metric_value));
        return APP_STATE_FAWPKT | APP_STATE_GIVEME;
    }
	//first data
	if(stream->ptcpdetail->datalen > 0){
		return first_data_process(stream, pmeinfo, &pktinfo, thread_seq);
	}
	//before first data, may be dup_syn, syn/ack, dup_syn/ack
	if(pktinfo.parse_failed != 0){
		KNI_LOG_ERROR(logger, "before first data: invalid ip header, bypass pkt");
		return APP_STATE_FAWPKT | APP_STATE_GIVEME;
	}
	tcp_handshake_pkt_process(pmeinfo, &pktinfo);
	return APP_STATE_FAWPKT | APP_STATE_GIVEME;
}

static int kni_set_policy_into_pem_info(const struct streaminfo *a_stream, struct pme_info *pmeinfo)
{
	struct traffic_info *traffic_info = &(pmeinfo->traffic_info);
	int value_len=sizeof(unsigned long long);
	if(a_stream == NULL ||  pmeinfo == NULL)
		return -1;
	traffic_info->con_num = 1;

	MESA_get_stream_opt(a_stream, MSO_TOTAL_INBOUND_BYTE_RAW, (void *)&traffic_info->in_bytes, &value_len);

	MESA_get_stream_opt(a_stream, MSO_TOTAL_INBOUND_PKT, (void *)&traffic_info->in_packets, &value_len);

	MESA_get_stream_opt(a_stream, MSO_TOTAL_OUTBOUND_BYTE_RAW, (void *)&traffic_info->out_bytes, &value_len);

	MESA_get_stream_opt(a_stream, MSO_TOTAL_OUTBOUND_PKT, (void *)&traffic_info->out_packets, &value_len);

	return 0;
}

static char close_opstate(const struct streaminfo *stream, struct pme_info *pmeinfo, int thread_seq){
	//close: because of timeout, return value has no meaning
	switch(pmeinfo->action){
		case KNI_ACTION_INTERCEPT:
			//TLD_append_streaminfo(g_tsg_log_instance, pmeinfo->tld_handle, (struct streaminfo*)pmeinfo->stream);
			kni_set_policy_into_pem_info(stream,pmeinfo);
			//reset clock: when sapp end, start clock
			// if(pmeinfo->is_dynamic_bypass != 1)
			// {
			// 	MESA_htable_search(g_kni_handle->traceid2pme_htable, (const unsigned char*)pmeinfo->stream_traceid,
			// 			strnlen(pmeinfo->stream_traceid, sizeof(pmeinfo->stream_traceid)));
			// 	tuple2stream_htable_del(stream, thread_seq);
			// }
            set_proxy_metric_value_by_packet(stream, &(pmeinfo->proxy_metric_value));
            if(pmeinfo->is_dynamic_bypass != 1 && pmeinfo->intcp_error == 0)
            {
                MESA_htable_search(g_kni_handle->traceid2pme_htable, (const unsigned char*)pmeinfo->stream_traceid,
                               strnlen(pmeinfo->stream_traceid, sizeof(pmeinfo->stream_traceid)));
                tuple2stream_htable_del(stream, thread_seq);
            }
			return APP_STATE_FAWPKT | APP_STATE_DROPME;
        case KNI_ACTION_NO_INTERCEPT:
            set_proxy_metric_value_by_packet(stream, &(pmeinfo->proxy_metric_value));
		//stream has no data.
		default:
			return APP_STATE_FAWPKT | APP_STATE_DROPME;
	}
}

static void pending_opstate(struct streaminfo *stream, struct pme_info *pmeinfo, const void *a_packet, int thread_seq){
	void *logger = g_kni_handle->local_logger;
	pme_info_init(pmeinfo, stream, thread_seq);
	struct pkt_info pktinfo;
	wrapped_kni_header_parse(a_packet, pmeinfo, &pktinfo);
	if(pktinfo.parse_failed == 1){
		KNI_LOG_ERROR(logger, "pending opstate: invalid ip header, bypass pkt");
		return;
	}
	tcp_handshake_pkt_process(pmeinfo, &pktinfo);
	return;	
}

/*sapp release: bypass or intercept
dropme has 3 status:
    0. Not hit intercept and no intercept maat rule.
       Action != KNI_ACTION_INTERCEPT
    1. Hit no intercept maat rule.
       Action == KNI_ACTION_INTERCEPT && maat_rule.action == TSG_ACTION_NO_INTERCEPT
    2. Hit intercept maat rule and intercept error.
       Action == KNI_ACTION_INTERCEPT && 
       maat_rule.action == TSG_ACTION_NO_INTERCEPT &&
       maat_rule.intercept_error < 0
    3. Hit intercept maat rule and  no intercept error.
       Action == KNI_ACTION_INTERCEPT && 
       maat_rule.action == TSG_ACTION_NO_INTERCEPT &&
       maat_rule.intercept_error == 0     
*/
static void deal_app_state_dropme(struct pme_info *pmeinfo, int thread_seq)
{
    int can_destroy = 0;

    if(pmeinfo == NULL)
    {
        return;
    }

    //not hit intercept and not intercept rule
    if(pmeinfo->action != KNI_ACTION_INTERCEPT && 
       pmeinfo->action != KNI_ACTION_NO_INTERCEPT)
    {
        stream_destroy(pmeinfo);
        return;
    }

    TLD_append_streaminfo(g_tsg_log_instance, pmeinfo->tld_handle,
                          (struct streaminfo*)pmeinfo->stream);
    //hit no intercept rule
    if(pmeinfo->action == KNI_ACTION_NO_INTERCEPT)
    {
        set_proxy_rule_hits_metric(pmeinfo, thread_seq);
        stream_destroy(pmeinfo);
        return;
    }

    //hit intercept rule and intercept error
    if(pmeinfo->action == KNI_ACTION_INTERCEPT && 
       pmeinfo->intcp_error < 0)
    {
        pmeinfo->ssl_intercept_state = 0;
        FS_operate(g_kni_fs_handle->handle,
                    g_kni_fs_handle->fields[KNI_FIELD_BYP_INTCPERR],
                    0, FS_OP_ADD, 1);
        set_proxy_rule_hits_metric(pmeinfo, thread_seq);
        stream_destroy(pmeinfo);
        return;
    }

     //hit intercept rule and no intercept error
    if(pmeinfo->action == KNI_ACTION_INTERCEPT &&
       pmeinfo->intcp_error == 0)
    {
        if(pmeinfo->is_dynamic_bypass == 0)
        {
            can_destroy = judge_stream_can_destroy(pmeinfo, CALLER_SAPP);
            if(can_destroy == 1)
            {
                traceid2pme_htable_del(pmeinfo);
                set_proxy_rule_hits_metric(pmeinfo, thread_seq);
                stream_destroy(pmeinfo);
            }
        }
        else
        {
            set_proxy_rule_hits_metric(pmeinfo, thread_seq);
            stream_destroy(pmeinfo);	
        }

    }
    return;
}

//from syn
extern "C" char kni_tcpall_entry(struct streaminfo *stream, void** pme, int thread_seq, const void* a_packet){
	void *logger = g_kni_handle->local_logger;
	int ret;
	struct pme_info *pmeinfo = *(struct pme_info **)pme;
	/*  a_packet == NULL && not op_state_close, continue
		close: a_packet may be null, if a_packet = null, do not send to tfe
	*/
	if(a_packet == NULL && stream->pktstate != OP_STATE_CLOSE){
		//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_NULL_PKT], 0, FS_OP_ADD, 1);
		return APP_STATE_FAWPKT | APP_STATE_GIVEME;
	}
	enum addr_type_t addr_type = (enum addr_type_t)stream->addr.addrtype;
	if(addr_type != ADDR_TYPE_IPV6 && addr_type != ADDR_TYPE_IPV4){
		KNI_LOG_ERROR(logger, "addr_type(%d) is not ipv4 or ipv6, bypass stream");
		return APP_STATE_FAWPKT | APP_STATE_DROPME;
	}

	switch(stream->pktstate){
		case OP_STATE_PENDING:
			//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_STATE_PENDING], 0, FS_OP_ADD, 1);
			pmeinfo = ALLOC(struct pme_info, 1);
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_PME_NEW_SUCC], 0, FS_OP_ADD, 1);
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_PME_CNT], 0, FS_OP_ADD, 1);
			*pme = pmeinfo;
			pending_opstate(stream, pmeinfo, a_packet, thread_seq);
			ret = APP_STATE_FAWPKT | APP_STATE_GIVEME;
			break;
		case OP_STATE_DATA:
			ret = data_opstate(stream, pmeinfo, a_packet, thread_seq);
			break;
		case OP_STATE_CLOSE:
			//sapp stream close
			//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_STATE_CLOSE], 0, FS_OP_ADD, 1);
			ret = close_opstate(stream, pmeinfo, thread_seq);
			break;
		default:
			ret = APP_STATE_FAWPKT | APP_STATE_GIVEME;
			//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_STATE_UNKNOWN], 0, FS_OP_ADD, 1);
			KNI_LOG_ERROR(logger, "Unknown stream opstate %d, stream traceid = %s, stream addr = %s", 
							stream->pktstate, pmeinfo->stream_traceid, pmeinfo->stream_addr);
			break;
	}

	if((ret & APP_STATE_DROPME))
    {
        deal_app_state_dropme(pmeinfo, thread_seq);
	}
	return ret;
}

static void kni_marsio_destroy(struct kni_marsio_handle *handle){
	if(handle != NULL){
		if(handle->instance != NULL){
			marsio_destory(handle->instance);
		}
	}
	FREE(&handle);
	handle = NULL;
}


int tuple2stream_htable_search(MESA_htable_handle handle, struct ethhdr *ether_hdr, int thread_seq){
	void *logger = g_kni_handle->local_logger;
	if(ether_hdr->h_proto != htons(ETH_P_IP) && ether_hdr->h_proto != htons(ETH_P_IPV6)){
		return -1;
	}
	void *raw_packet = (char*)ether_hdr + sizeof(*ether_hdr);
	tuple2stream_htable_value *value = NULL;
	struct pkt_info pktinfo;
	int reversed = 0, ret;
	char key_str[KNI_ADDR_MAX];
	struct stream_tuple4_v6 key_v6;
	struct stream_tuple4_v4 key_v4;
	memset(key_str,0,sizeof(key_str));
	//ipv6
	if(ether_hdr->h_proto == htons(ETH_P_IPV6)){
		ret = kni_ipv6_header_parse(raw_packet, &pktinfo);
		if(ret < 0){
			char *errmsg = kni_ipv6_errmsg_get((enum kni_ipv6hdr_parse_error)ret);
			KNI_LOG_ERROR(logger, "failed at parse ipv6 header, errmsg = %s", errmsg);
			//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_IPV6HDR_PARSE_FAIL], 0, FS_OP_ADD, 1);
			return -1;
		}
		tuple2stream_htable_key_get_v6_by_packet(&pktinfo, &key_v6, &reversed);
		value = (tuple2stream_htable_value*)MESA_htable_search(handle, (const unsigned char*)(&key_v6), sizeof(key_v6));
	}
	//ipv4
	else{
		ret = kni_ipv4_header_parse(raw_packet, &pktinfo);
		if(ret < 0){
			char *errmsg = kni_ipv4_errmsg_get((enum kni_ipv4hdr_parse_error)ret);
			KNI_LOG_ERROR(logger, "failed at parse ipv4 header, errmsg = %s", errmsg);
			//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_IPV4HDR_PARSE_FAIL], 0, FS_OP_ADD, 1);
			return -1;
		}
		tuple2stream_htable_key_get_v4_by_packet(&pktinfo, &key_v4, &reversed);
		value = (tuple2stream_htable_value*)MESA_htable_search(handle, (const unsigned char*)(&key_v4), sizeof(key_v4));
	}
	if(value == NULL){
		if(ether_hdr->h_proto == htons(ETH_P_IPV6)){
			kni_addr_trans_v6(&key_v6, key_str, sizeof(key_str));
		}
		else{
			kni_addr_trans_v4(&key_v4, key_str, sizeof(key_str));
		}
		KNI_LOG_DEBUG(logger, "MESA_htable: search not hit, table is tuple2stream_htable, key = %s, key_size = %d", key_str, strlen(key_str));
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_TUPLE2STM_SEARCH_MISS], 0, FS_OP_ADD, 1);
		return -1;
	}
	//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_TUPLE2STM_SEARCH_HIT], 0, FS_OP_ADD, 1);
	unsigned char dir = value->route_dir;
	if(reversed != value->reversed){
		dir = MESA_dir_reverse(dir);
	}

	ret = sapp_inject_pkt(value->stream, SIO_EXCLUDE_THIS_LAYER_HDR, raw_packet, pktinfo.ip_totlen, dir);
	if(ret < 0){
		if(ether_hdr->h_proto == htons(ETH_P_IPV6)){
			kni_addr_trans_v6(&key_v6, key_str, sizeof(key_str));
		}
		else{
			kni_addr_trans_v4(&key_v4, key_str, sizeof(key_str));
		}
		KNI_LOG_ERROR(logger, "Failed at sapp_inject_pkt, stream addr = %s", key_str);
		FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_SAPP_INJECT_FAIL], 0, FS_OP_ADD, 1);
		return -1;
	}
	//FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_SAPP_INJECT_SUCC], 0, FS_OP_ADD, 1);
	//add to dabloom
	if(g_kni_handle->dup_traffic_switch == 1){
		if(value->pmeinfo->has_dup_traffic == 1){ 
			ret = dabloom_add(&pktinfo, thread_seq);
			if(ret < 0){
				return -1;
			}
		}
	}
	return 0;
}

extern "C" char kni_polling_all_entry(const struct streaminfo *stream, void** pme, int thread_seq, const void* a_packet){
	void *logger = g_kni_handle->local_logger;
	MESA_htable_handle tuple2stream_htable = g_kni_handle->threads_handle[thread_seq].tuple2stream_htable;
	int flag = POLLING_STATE_IDLE;
	//normal mode
	if(g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_NORMAL){
		//polling tfe
		for(int i = 0; i < g_kni_handle->marsio_handle->tfe_enabled_node_count; i++){
			marsio_buff_t *rx_buffs[BURST_MAX];
			int nr_burst = 1;
			struct mr_vdev *dev_eth_handler = g_kni_handle->marsio_handle->tfe_enabled_nodes[i].dev_eth_handler;
			//receive from tfe, nr_recv <= nr_burst <= BURST_MAX
			int nr_recv = marsio_recv_burst(dev_eth_handler, thread_seq, rx_buffs, nr_burst);
			if(nr_recv <= 0){
				continue;
			}
			for(int j = 0; j < nr_recv; j++){
				struct ethhdr *ether_hdr = (struct ethhdr*)marsio_buff_mtod(rx_buffs[j]);
				tuple2stream_htable_search(tuple2stream_htable, ether_hdr, thread_seq);
				flag = POLLING_STATE_WORK;
			}
			marsio_buff_free(g_kni_handle->marsio_handle->instance, rx_buffs, nr_recv, 0, 0);
		}
	}
	else if (g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_TAP)
	{
		int tap_fd = g_kni_handle->threads_handle[thread_seq].tap_fd;
		if (g_kni_handle->iouring_conf.enable_iouring)
		{
			if (io_uring_polling(thread_seq) == -1)
			{
				flag = POLLING_STATE_IDLE;
			}
			else
			{
				flag = POLLING_STATE_WORK;
			}
		}
		else
		{
			char *recv_buff = g_kni_handle->threads_handle[thread_seq].buff;
			int recv_buff_size = g_kni_handle->threads_handle[thread_seq].buff_size;

			// 只支持一个 TFE, 每次最多收 100 个包
			for (int j = 0; j < 100; j++)
			{
				if (kni_tap_read_per_thread(tap_fd, recv_buff, recv_buff_size, logger) > 0)
				{
					struct ethhdr *ether_hdr = (struct ethhdr *)recv_buff;
					tuple2stream_htable_search(tuple2stream_htable, ether_hdr, thread_seq);
					flag = POLLING_STATE_WORK;
				}
				else
				{
					break;
				}
			}
		}
	}

	return flag;
}


static int wrapped_kni_cmsg_get(struct pme_info *pmeinfo, struct kni_cmsg *cmsg, uint16_t type, 
			uint16_t value_size_max, void *logger){
	uint16_t value_size = 0;
	unsigned char *value = NULL;
	int ret = kni_cmsg_get(cmsg, type, &value_size, &value);
	if(ret < 0){
		if(ret == KNI_CMSG_INVALID_TYPE){
			KNI_LOG_ERROR(logger, "Failed at kni_cmsg_get: type = %d, ret = %d, stream traceid = %s, stream addr = %s",
					 type, ret, pmeinfo->stream_traceid, pmeinfo->stream_addr);
		}
		return -1;
	}
	if(value_size > value_size_max){
		KNI_LOG_ERROR(logger, "kni_cmsg_get: type = %d, size = %d, which should <= %d, stream traceid = %s, stream addr = %s",
					 type, value_size, value_size_max, pmeinfo->stream_traceid, pmeinfo->stream_addr);
		return -1;
	}
	switch(type)
	{
		case TFE_CMSG_SSL_INTERCEPT_STATE:
			memcpy((char*)&(pmeinfo->ssl_intercept_state), value, value_size);
			break;
		case TFE_CMSG_SSL_UPSTREAM_LATENCY:
			memcpy((char*)&(pmeinfo->ssl_server_side_latency), value, value_size);
			break;
		case TFE_CMSG_SSL_DOWNSTREAM_LATENCY:
			memcpy((char*)&(pmeinfo->ssl_client_side_latency), value, value_size);
			break;
		case TFE_CMSG_SSL_UPSTREAM_VERSION:
			memcpy(pmeinfo->ssl_server_side_version, value, value_size);
			break;
		case TFE_CMSG_SSL_DOWNSTREAM_VERSION:
			memcpy(pmeinfo->ssl_client_side_version, value, value_size);
			break;
		case TFE_CMSG_SSL_PINNING_STATE:
			memcpy((char*)&(pmeinfo->ssl_pinningst), value, value_size);
			break;
		case TFE_CMSG_SSL_CERT_VERIFY:
			memcpy((char*)&(pmeinfo->ssl_cert_verify), value, value_size);
			break;
		case TFE_CMSG_SSL_ERROR:
			memcpy((char*)&(pmeinfo->ssl_error), value, value_size);
			break;
		case TFE_CMSG_SSL_PASSTHROUGH_REASON:
			memcpy(pmeinfo->ssl_passthrough_reason, value, value_size);
			break;
		default:
			break;
	}
	return 0;
}


static long traceid2pme_htable_search_cb(void *data, const uchar *key, uint size, void *user_args){
	struct traceid2pme_search_cb_args *args = (struct traceid2pme_search_cb_args*)user_args;
	void *logger = args->logger;
	struct kni_cmsg *cmsg = args->cmsg;
	struct pme_info *pmeinfo = (struct pme_info*)data;
	int can_destroy;
	if(pmeinfo != NULL){
		wrapped_kni_cmsg_get(pmeinfo, cmsg, TFE_CMSG_SSL_INTERCEPT_STATE, sizeof(pmeinfo->ssl_intercept_state), logger);
		wrapped_kni_cmsg_get(pmeinfo, cmsg, TFE_CMSG_SSL_UPSTREAM_LATENCY, sizeof(pmeinfo->ssl_server_side_latency), logger);
		wrapped_kni_cmsg_get(pmeinfo, cmsg, TFE_CMSG_SSL_DOWNSTREAM_LATENCY, sizeof(pmeinfo->ssl_client_side_latency), logger);
		wrapped_kni_cmsg_get(pmeinfo, cmsg, TFE_CMSG_SSL_UPSTREAM_VERSION, sizeof(pmeinfo->ssl_server_side_version) - 1, logger);
		wrapped_kni_cmsg_get(pmeinfo, cmsg, TFE_CMSG_SSL_DOWNSTREAM_VERSION, sizeof(pmeinfo->ssl_client_side_version) - 1, logger);
		wrapped_kni_cmsg_get(pmeinfo, cmsg, TFE_CMSG_SSL_PINNING_STATE, sizeof(pmeinfo->ssl_pinningst), logger);
		wrapped_kni_cmsg_get(pmeinfo, cmsg, TFE_CMSG_SSL_CERT_VERIFY, sizeof(pmeinfo->ssl_cert_verify), logger);
		wrapped_kni_cmsg_get(pmeinfo, cmsg, TFE_CMSG_SSL_ERROR, sizeof(pmeinfo->ssl_error), logger);
		wrapped_kni_cmsg_get(pmeinfo, cmsg, TFE_CMSG_SSL_PASSTHROUGH_REASON, sizeof(pmeinfo->ssl_passthrough_reason), logger);
		KNI_LOG_DEBUG(logger, "recv cmsg from tfe, stream traceid = %s, stream addr = %s, stream ssl intercept state = %d ,pinning state = %d", 
							pmeinfo->stream_traceid, pmeinfo->stream_addr,pmeinfo->ssl_intercept_state,pmeinfo->ssl_pinningst);

		if(g_kni_handle->ssl_dynamic_bypass_enable == 1){
			ssl_dynamic_bypass_htable_add(pmeinfo);
		}

		can_destroy = judge_stream_can_destroy(pmeinfo, CALLER_TFE);
		if(can_destroy == 1){
			traceid2pme_htable_del(pmeinfo);
			stream_destroy(pmeinfo);
		}
	}
	kni_cmsg_destroy(cmsg);
	return 0;
}

static void* thread_tfe_cmsg_receiver(void *args){
	struct thread_tfe_cmsg_receiver_args *_args = (struct thread_tfe_cmsg_receiver_args*)args;
	const char *profile = _args->profile;
	const char *section = "tfe_cmsg_receiver";
	void *logger = _args->logger;
	char listen_eth[KNI_SYMBOL_MAX];
	uint32_t listen_ip;
	int listen_port = -1;
	char buff[KNI_MTU];
	int sockfd = 0;
	struct sockaddr_in server_addr, client_addr;
	int ret = MESA_load_profile_string_nodef(profile, section, "listen_eth", listen_eth, sizeof(listen_eth));
	if(ret < 0){
		KNI_LOG_ERROR(logger, "MESA_prof_load: listen_eth not set, profile = %s, section = %s", profile, section);
		goto error_out;
	}
	ret = MESA_load_profile_int_nodef(profile, section, "listen_port", &listen_port);
	if(ret < 0){
		KNI_LOG_ERROR(logger, "MESA_prof_load: listen_port not set, profile = %s, section = %s", profile, section);
		goto error_out;
	}
	KNI_LOG_ERROR(logger, "MESA_prof_load, [%s]:\n listen_eth: %s\n listen_port: %d",
					section, listen_eth, listen_port);
	FREE(&args);
	//create socket
	sockfd = socket(AF_INET, SOCK_DGRAM, 0);
	if(sockfd < 0){
		KNI_LOG_ERROR(logger, "Failed at create udp socket, errno = %d, %s", errno, strerror(errno));
		goto error_out;
	}
	memset(&server_addr, 0, sizeof(server_addr)); 
    memset(&client_addr, 0, sizeof(client_addr)); 
	ret = kni_ipv4_addr_get_by_eth(listen_eth, &listen_ip);
	if(ret < 0){
		KNI_LOG_ERROR(logger, "Failed at get bind ipv4 addr, eth = %s", listen_eth);
		goto error_out;
	}
	server_addr.sin_family = AF_INET; // IPv4 
    server_addr.sin_addr.s_addr = listen_ip; 
    server_addr.sin_port = htons(listen_port);
	//bind
	ret = bind(sockfd, (const struct sockaddr *)&server_addr, sizeof(server_addr));
	if(ret < 0){
		KNI_LOG_ERROR(logger, "Failed at bind udp socket, errno = %d, %s", errno, strerror(errno));
		goto error_out;
	}
	//receive
	while(true){
		socklen_t client_len = sizeof(client_addr);
		int recv_len = recvfrom(sockfd, (char *)buff, sizeof(buff), MSG_WAITALL,
			 (struct sockaddr*)&client_addr, &client_len);
		if(recv_len < 0){
			KNI_LOG_ERROR(logger, "Failed at recv udp data, errno = %d, %s", errno, strerror(errno));
			continue;
		}
		//KNI_LOG_DEBUG(logger, "recv udp data: recv_len = %d\n", recv_len);
		struct kni_cmsg *cmsg = NULL;
		ret = kni_cmsg_deserialize((const unsigned char*)buff, recv_len, &cmsg);
		if(ret < 0){
			KNI_LOG_ERROR(logger, "Failed at deserialize cmsg, ret = %d", ret);
			continue;
		}
		//get stream_traceid
		unsigned char *stream_traceid = NULL;
		uint16_t value_size;
		ret = kni_cmsg_get(cmsg, TFE_CMSG_STREAM_TRACE_ID, &value_size, &stream_traceid);
		if(ret < 0){
			KNI_LOG_ERROR(logger, "Failed at kni_cmsg_get: type = %d, ret = %d", TFE_CMSG_STREAM_TRACE_ID, ret);
			continue;
		}
		//get pme
		long cb_ret = -1;
		struct traceid2pme_search_cb_args cb_args;
		memset((void*)&cb_args, 0, sizeof(cb_args));
		cb_args.cmsg = cmsg;
		cb_args.logger = logger;
        MESA_htable_search_cb(g_kni_handle->traceid2pme_htable, (const unsigned char *)stream_traceid, 
			value_size, traceid2pme_htable_search_cb, &cb_args, &cb_ret);
	}
	return NULL;

error_out:
	if(sockfd >= 0){
		close(sockfd);
	}
	return NULL;
}

static struct kni_marsio_handle* kni_marsio_init(const char* profile, int tfe_node_count){
	void *logger = g_kni_handle->local_logger;
	const char* section = "marsio";
	char appsym[KNI_SYMBOL_MAX];
	unsigned int opt_value = 1;
	int tfe_node_enabled;
	struct mr_instance *mr_inst = NULL;
	struct mr_vdev *dev_eth_handler = NULL;
	struct mr_sendpath *dev_eth_sendpath = NULL;
	struct kni_marsio_handle *handle = NULL;
	int j;
	int ret = MESA_load_profile_string_nodef(profile, section, "appsym", appsym, sizeof(appsym));
	if(ret < 0){
		KNI_LOG_ERROR(logger, "MESA_prof_load: appsym not set, profile = %s, section = %s", profile, section);
		goto error_out;
	}
	KNI_LOG_ERROR(logger, "MESA_prof_load, [%s]:\n appsym: %s", section, appsym);
	mr_inst = marsio_create();
	if(mr_inst == NULL){
		KNI_LOG_ERROR(logger, "Failed at create marsio instance");
		goto error_out;
	}
	handle = ALLOC(struct kni_marsio_handle, 1);
	handle->instance = mr_inst;
    marsio_option_set(mr_inst, MARSIO_OPT_EXIT_WHEN_ERR, &opt_value, sizeof(opt_value));
    marsio_init(mr_inst, appsym);
	j = 0;
	for(int i = 0; i < tfe_node_count; i++){
		//load tfe conf
		char _section[KNI_SYMBOL_MAX];
		char dev_eth_symbol[KNI_SYMBOL_MAX];
		snprintf(_section, sizeof(_section), "tfe%d", i);
		MESA_load_profile_int_def(profile, _section, "enabled", &tfe_node_enabled, 1);
		if(tfe_node_enabled != 1){
			continue;
		}
		struct tfe_enabled_node tfe_node;
		memset(&tfe_node, 0, sizeof(tfe_node));
		ret = MESA_load_profile_string_nodef(profile, _section, "dev_eth_symbol", dev_eth_symbol, sizeof(dev_eth_symbol));
		if(ret < 0){
			KNI_LOG_ERROR(logger, "MESA_prof_load: dev_eth_symbol not set, profile = %s, section = %s", profile, _section);
			goto error_out;
		}
		KNI_LOG_ERROR(logger, "MESA_prof_load, [%s]:\n enabled: %d\n dev_eth_symbol: %s",
					_section, tfe_node_enabled, dev_eth_symbol);
		//eth_handler receive thread = thread_count, send thread = thread_count
		dev_eth_handler = marsio_open_device(mr_inst, dev_eth_symbol, g_kni_handle->thread_count, g_kni_handle->thread_count);
		if(dev_eth_handler == NULL){
			KNI_LOG_ERROR(logger, "Failed at marsio_open_device, dev_symbol = %s", dev_eth_symbol);
			goto error_out;
		}
		//sendpath
		dev_eth_sendpath = marsio_sendpath_create_by_vdev(dev_eth_handler);
		if(dev_eth_sendpath == NULL){
			KNI_LOG_ERROR(logger, "Failed at create marsio sendpath, dev_symbol = %s", dev_eth_symbol);
			goto error_out;
		}
		//tfe_node
		tfe_node.dev_eth_handler = dev_eth_handler;
		tfe_node.dev_eth_sendpath = dev_eth_sendpath;
		tfe_node.tfe_id = i;
		handle->tfe_enabled_nodes[j++] = tfe_node;
	}
	handle->tfe_enabled_node_count = j;
	//marsio_thread_init(mr_instance); 
	return handle;

error_out:
	kni_marsio_destroy(handle);
	return NULL;
}

static void fs_destroy(struct kni_field_stat_handle *fs_handle){
	if(fs_handle != NULL){
		FS_stop(&(fs_handle->handle));
	}
	FREE(&fs_handle);
}

static struct kni_field_stat_handle * fs_init(const char *profile){
	void *logger = g_kni_handle->local_logger;
	const char *section = "field_stat";
	char local_path[KNI_PATH_MAX];
	struct kni_field_stat_handle *fs_handle = NULL;
	screen_stat_handle_t handle = NULL;
	char app_name[MAX_STRING_LEN]={0};
	int value = 0, ret, stat_cycle, print_mode;
	int remote_switch = 0;
	char remote_ip[INET_ADDRSTRLEN];
	int remote_port;
	int statsd_format = FS_OUTPUT_STATSD;
	MESA_load_profile_string_def(profile, section, "APP_NAME", app_name, sizeof(app_name), "fs2_kni");
	MESA_load_profile_int_def(profile, section, "remote_switch", &remote_switch, 0);
	MESA_load_profile_string_def(profile, section, "local_path", local_path, sizeof(local_path), "./fs2_kni.status");
	KNI_LOG_ERROR(logger, "MESA_prof_load, [%s]:\n remote_switch: %d\n local_path: %s", section, remote_switch, local_path);
	handle = FS_create_handle();
	if(handle == NULL){
		KNI_LOG_ERROR(logger, "Failed at create FS_create_handle");
		goto error_out;
	}
	if(remote_switch == 1){
		ret = MESA_load_profile_string_nodef(profile, section, "remote_ip", remote_ip, sizeof(remote_ip));
		if(ret < 0){
			KNI_LOG_ERROR(logger, "MESA_prof_load: remote_ip not set, profile is %s, section is %s", profile, section);
			goto error_out;
		}
		ret = MESA_load_profile_int_nodef(profile, section, "remote_port", &remote_port);
		if(ret < 0){
			KNI_LOG_ERROR(logger, "MESA_prof_load: remote_port not set, profile is %s, section is %s", profile, section);
			goto error_out;
		}
		KNI_LOG_ERROR(logger, "MESA_prof_load, [%s]:\n remote_ip: %s\n remote_port: %d", section, remote_ip, remote_port);
		FS_set_para(handle, STATS_SERVER_IP, remote_ip, strlen(remote_ip));
		FS_set_para(handle, STATS_SERVER_PORT, &remote_port, sizeof(remote_port));
		MESA_load_profile_int_def(profile, section, "statsd_format", &statsd_format, 0);
		switch(statsd_format)
		{
			case 1:
				value=FS_OUTPUT_STATSD;
				break;
			case 2:
				value=FS_OUTPUT_INFLUX_LINE;
				break;
			default:
				value=FS_OUTPUT_STATSD;
		}
		FS_set_para(handle, STATS_FORMAT, &value, sizeof(value));
	}
	MESA_load_profile_int_def(profile, section, "stat_cycle", &stat_cycle, 5);
	MESA_load_profile_int_def(profile, section, "print_mode", &print_mode, 1);
	fs_handle = ALLOC(struct kni_field_stat_handle, 1);
	fs_handle->handle = handle;
	FS_set_para(handle, APP_NAME, app_name, strlen(app_name) + 1);
	FS_set_para(handle, OUTPUT_DEVICE, local_path, strlen(local_path)+1);
	value = 0;
	FS_set_para(handle, FLUSH_BY_DATE, &value, sizeof(value));
	value = print_mode;
	FS_set_para(handle, PRINT_MODE, &value, sizeof(value));
	value = 1;
	FS_set_para(handle, CREATE_THREAD, &value, sizeof(value));
	value = stat_cycle;
	FS_set_para(handle, STAT_CYCLE, &value, sizeof(value));
	value = 4096;
	FS_set_para(handle, MAX_STAT_FIELD_NUM, &value, sizeof(value));

	value=1;
	FS_set_para(handle, OUTPUT_PROMETHEUS, &value, sizeof(value));

	fs_handle = ALLOC(struct kni_field_stat_handle, 1);
	fs_handle->handle = handle;

	//bypass stream
	fs_handle->fields[KNI_FIELD_BYP_INTCPERR] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "byp_intcp_err");
	//intercept error  link mode
	fs_handle->fields[KNI_FIELD_INTCPERR_GET_LINK_MODE_ERR] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_lkmd_get");
	fs_handle->fields[KNI_FIELD_INTCPERR_NOT_LINK_MODE_BYSYN] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_lkmd_not_syn");
    //intercept error  get HAVE_DUP_PKT error
	fs_handle->fields[KNI_FIELD_INTCPERR_GET_HAVE_DUP_PKT_ERR] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_dup_get");
	fs_handle->fields[KNI_FIELD_INTCPERR_DUP_PKT_NOT_SURE_ERR] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_dup_notsure");
	//intercept error  stream tunnel type
	fs_handle->fields[KNI_FIELD_INTCPERR_GET_STREAM_TUN_TYPE_ERR] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_tuntype_get");
	fs_handle->fields[KNI_FIELD_INTCPERR_STREAM_IS_TUN_TYPE] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_type_tun");	
	//intercept_error
	fs_handle->fields[KNI_FIELD_INTCPERR_ASYM_ROUTING] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_asym_route");
	fs_handle->fields[KNI_FIELD_INTCPERR_NO_SYN] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_no_syn");
	fs_handle->fields[KNI_FIELD_INTCPERR_NO_SYN_ACK] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_no_s/a");
	fs_handle->fields[KNI_FIELD_INTCPERR_INVALID_IP_HDR] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_ip_hdr");
	fs_handle->fields[KNI_FIELD_INTCPERR_EXCEED_MTU] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_exc_mtu");
	//intercept_error: internal error
	fs_handle->fields[KNI_FIELD_INTCPERR_SENDTO_TFE_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_tfe_tx");
	fs_handle->fields[KNI_FIELD_INTCPERR_TUPLE2STM_ADD_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_tup2stm_add");
	fs_handle->fields[KNI_FIELD_INTCPERR_NO_TFE] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_no_tfe");
	fs_handle->fields[KNI_FIELD_INTCPERR_DUP_TRAFFIC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_dup_tfc");
	fs_handle->fields[KNI_FIELD_INTCPERR_CMSG_ADD_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_cmsg_add");

	//success intercept stream
	fs_handle->fields[KNI_FIELD_INTCP_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "intcp_stm");
	//fs_handle->fields[KNI_FIELD_INTCP_BYTE] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "intcp_B");
	fs_handle->fields[KNI_FIELD_IPV4_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "ipv4_stm");
	fs_handle->fields[KNI_FIELD_IPV6_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "ipv6_stm");
	fs_handle->fields[KNI_FIELD_SSL_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "ssl_stm");
	fs_handle->fields[KNI_FIELD_HTTP_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "http_stm");
	fs_handle->fields[KNI_FIELD_DUP_TFC_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "dup_tfc_stm");
	fs_handle->fields[KNI_FIELD_DUP_TFC_BYTE] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "dup_tfc_B");

    fs_handle->fields[KNI_FIELD_NO_INTCP_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "no_intcp_rdy_stm");
	//intercept ready stream:  success + failed 
	fs_handle->fields[KNI_FIELD_INTCP_READY_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "intcp_rdy_stm");
	//fs_handle->fields[KNI_FIELD_INTCP_READY_BYTE] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "intcp_rdy_B");

	//pme 
	fs_handle->fields[KNI_FIELD_PME_NEW_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "pme_new");
	fs_handle->fields[KNI_FIELD_PME_FREE] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "pme_free");
	fs_handle->fields[KNI_FIELD_PME_CNT] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "pme_cnt");

	//errors
	fs_handle->fields[KNI_FIELD_SENDLOG_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_sendlog");
	fs_handle->fields[KNI_FIELD_ID2PME_ADD_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_id2pme_add");
	fs_handle->fields[KNI_FIELD_ID2PME_DEL_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_id2pme_del");
	fs_handle->fields[KNI_FIELD_TUPLE2STM_ADD_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_tup2stm_add");
	fs_handle->fields[KNI_FIELD_TUPLE2STM_DEL_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_tup2stm_del");
	fs_handle->fields[KNI_FIELD_SAPP_INJECT_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_sapp_inject");
	fs_handle->fields[KNI_FIELD_BLOOM_SEARCH_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_bloom_srch");
	fs_handle->fields[KNI_FIELD_BLOOM_ADD_FAIL] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "e_bloom_add");

	//htable
	fs_handle->fields[KNI_FIELD_ID2PME_ADD_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "id2pme_add_S");
	fs_handle->fields[KNI_FIELD_ID2PME_DEL_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "id2pme_del_S");
	fs_handle->fields[KNI_FIELD_ID2PME_CNT] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "id2pme_cnt");
	fs_handle->fields[KNI_FIELD_TUPLE2STM_ADD_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "tup2stm_add_S");
	fs_handle->fields[KNI_FIELD_TUPLE2STM_DEL_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "tup2stm_del_S");
	//fs_handle->fields[KNI_FIELD_TUPLE2STM_SEARCH_HIT] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "tup2stm_hit");
	fs_handle->fields[KNI_FIELD_TUPLE2STM_SEARCH_MISS] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "tup2stm_miss");
	//sendlog
	fs_handle->fields[KNI_FIELD_SENDLOG_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "sendlog_S");

	//sapp_inject
	//fs_handle->fields[KNI_FIELD_SAPP_INJECT_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "sapp_inject_S");

	//dabloom
	fs_handle->fields[KNI_FIELD_BLOOM_HIT] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "bloom_hit");
	fs_handle->fields[KNI_FIELD_BLOOM_MISS] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "bloom_miss");

	//dynamic bypass
	fs_handle->fields[KNI_FIELD_ID2SSL_ADD_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "id2ssl_add_S");
	fs_handle->fields[KNI_FIELD_ID2SSL_DEL_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "id2ssl_del_S");
	fs_handle->fields[KNI_FIELD_ID2SSL_CNT] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "id2ssl_cnt");
	fs_handle->fields[KNI_FIELD_SSL2PASS_ADD_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "ssl2pass_add_S");
	fs_handle->fields[KNI_FIELD_SSL2PASS_DEL_SUCC] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "ssl2pass_del_S");
	fs_handle->fields[KNI_FIELD_SSL2PASS_CNT] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "ssl2pass_cnt");
	fs_handle->fields[KNI_FIELD_DY_PASS_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "dy_pass_stm");
	fs_handle->fields[KNI_FIELD_DY_PASS_BYTE] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "dy_pass_B");
	fs_handle->fields[KNI_FIELD_DY_PASS_IPV6_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "dy_pass_ipv6_stm");
	fs_handle->fields[KNI_FIELD_DY_PASS_IPV4_STM] = FS_register(handle, FS_STYLE_FIELD, FS_CALC_CURRENT, "dy_pass_ipv4_stm");


	if(g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_NORMAL){
		for(int i = 0; i < g_kni_handle->marsio_handle->tfe_enabled_node_count; i++){
			int tfe_id = g_kni_handle->marsio_handle->tfe_enabled_nodes[i].tfe_id;
			char tfe_status[KNI_SYMBOL_MAX] = "";
			snprintf(tfe_status, sizeof(tfe_status), "tfe%d", tfe_id);
			fs_handle->fields[KNI_FIELD_TFE_STATUS_BASE + i] = FS_register(handle, FS_STYLE_STATUS, FS_CALC_CURRENT, tfe_status);
		}
	}
	//table
	char buff[KNI_PATH_MAX];
	for(int i = 0; i < g_kni_handle->thread_count; i++){
		snprintf(buff, sizeof(buff), "tid%d", i);
		fs_handle->column_ids[i] = FS_register(handle, FS_STYLE_COLUMN, FS_CALC_CURRENT, buff);
	}
	snprintf(buff, sizeof(buff), "Total");
	//lines
	fs_handle->column_ids[g_kni_handle->thread_count] = FS_register(handle, FS_STYLE_COLUMN, FS_CALC_CURRENT, buff);
	snprintf(buff, sizeof(buff), "bloom_cnt");
	fs_handle->line_ids[0] = FS_register(handle, FS_STYLE_LINE, FS_CALC_CURRENT, buff);
	snprintf(buff, sizeof(buff), "tuple2stm_cnt");
	fs_handle->line_ids[1] = FS_register(handle, FS_STYLE_LINE, FS_CALC_CURRENT, buff);
	
	fs_handle->handle = handle;
	FS_start(handle);
	return fs_handle;

error_out:
	fs_destroy(fs_handle);
	return NULL;
}

extern "C" void kni_destroy()
{
	if (g_kni_handle != NULL)
	{
		for (int i = 0; i < g_kni_handle->thread_count; i++)
		{
			kni_tap_close_per_thread(g_kni_handle->threads_handle[i].tap_fd);

			if (g_kni_handle->threads_handle[i].buff)
			{
				free(g_kni_handle->threads_handle[i].buff);
				g_kni_handle->threads_handle[i].buff = NULL;
			}
		}

		kni_tap_global_unload_rss_bpf(g_kni_handle->tap_bpf_ctx);

		FREE(&g_kni_handle);
		g_kni_handle = NULL;
	}
}

//eliminate_type: 0:FIFO; 1:LRU
//ret: 1: the item can be eliminated; 0: the item can't be eliminated
static int traceid2pme_htable_expire_notify_cb(void *data, int eliminate_type){
	struct pme_info *pmeinfo = (struct pme_info*)data;
	int can_destroy;
	if(pmeinfo->sapp_release == 1){
		can_destroy = judge_stream_can_destroy(pmeinfo, CALLER_TFE);
		if(can_destroy == 1){
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_ID2PME_DEL_SUCC], 0, FS_OP_ADD, 1);
			FS_operate(g_kni_fs_handle->handle, g_kni_fs_handle->fields[KNI_FIELD_ID2PME_CNT], 0, FS_OP_ADD, -1);
            set_proxy_rule_hits_metric(pmeinfo, g_kni_handle->thread_count);
			stream_destroy(pmeinfo);
			return 1;
		}
	}
	return 0;
}

static void tuple2stream_htable_data_free_cb(void *data){
	FREE(&data);
}



int dup_traffic_dabloom_init(const char *profile, void *logger){
	const char *section = "dup_traffic";
	MESA_load_profile_int_def(profile, section, "switch", &(g_kni_handle->dup_traffic_switch), 0);
	KNI_LOG_ERROR(logger, "MESA_prof_load, [%s]:\n switch: %d", section, g_kni_handle->dup_traffic_switch);
	if(g_kni_handle->dup_traffic_switch == 1){	
		unsigned int capacity = 0;
		char error_rate_str[KNI_SYMBOL_MAX];
		double error_rate = 0.05;
		int expiry_time = 0;
		MESA_load_profile_int_def(profile, section, "action", &(g_kni_handle->dup_traffic_action), KNI_ACTION_BYPASS);
		MESA_load_profile_uint_def(profile, section, "capacity", &capacity, 1000000);
    	MESA_load_profile_string_def(profile, section, "error_rate", error_rate_str, sizeof(error_rate_str), "0.05");
    	MESA_load_profile_int_def(profile, section, "expiry_time", &expiry_time, 30);
		KNI_LOG_ERROR(logger, "MESA_prof_load, [%s]:\n action: %d\n capacity: %d\n error_rate: %s\n expiry_time: %d",
					section, g_kni_handle->dup_traffic_action, capacity, error_rate_str, expiry_time);
		error_rate = atof(error_rate_str);
		for(int i = 0; i < g_kni_handle->thread_count; i++){
			struct expiry_dablooms_handle* dabloom_handle = expiry_dablooms_init(capacity, error_rate, expiry_time);
			if(dabloom_handle == NULL){
				KNI_LOG_ERROR(logger, "Failed at expiry_dablooms_init, capacity = %d,"
							"error_rate = %lf, expire_time = %d", capacity, error_rate, expiry_time);
				return -1;
			}
			g_kni_handle->threads_handle[i].dabloom_handle = dabloom_handle;
		}
		return 0;
	}
	return 0;
}

/*
void my_handler(int s){
	printf("Caught signal %d\n",s);	
	exit(1);
}
int register_signal_handle(){
	struct sigaction sigIntHandler;
	sigIntHandler.sa_handler = my_handler;
	sigemptyset(&sigIntHandler.sa_mask);
	sigIntHandler.sa_flags = 0;
	sigaction(SIGINT, &sigIntHandler, NULL);
	return 0;
}
*/

int set_rps(void *local_logger, const char *tap_name, int thread_num, const char *rps_mask)
{
	char file[1024] = {0};
	for (int i = 0; i < thread_num; i++)
	{
		memset(file, 0, sizeof(file));
		snprintf(file, sizeof(file), "/sys/class/net/%s/queues/rx-%d/rps_cpus", tap_name, i);

		FILE *fp = fopen(file, "w");
		if (fp == NULL)
		{
			KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG, "can't open %s, %s", file, strerror(errno));
			return -1;
		}

		fwrite(rps_mask, strlen(rps_mask), 1, fp);
		KNI_LOG_DEBUG(local_logger, TAP_RSS_LOG_TAG "set rps '%s' to %s", rps_mask, file);
		fclose(fp);
	}
	return 0;
}

extern "C" int kni_init(){
	//register_signal_handle();
	char *kni_git_verison = (char*)KNI_GIT_VERSION;
	const char *profile = "./etc/kni/kni.conf";
	const char *section = "global";
	char tap_name[1024] = {0};
	//init logger
	char log_path[KNI_PATH_MAX] = "";
	int tfe_node_count = 1;
	char manage_eth[KNI_SYMBOL_MAX] = "";
	struct kni_field_stat_handle *fs_handle = NULL;
	void *local_logger = NULL;
	int log_level = -1;
	pthread_t thread_id = -1;
	struct thread_tfe_cmsg_receiver_args *cmsg_receiver_args;
	MESA_htable_handle traceid2pme_htable = NULL;
	struct tfe_mgr *_tfe_mgr = NULL;
	int n_pxy_fs_thread = 0;
	//char label_buff[MAX_STRING_LEN*4]={0};
	tfe_cmsg_enum_to_string();
	int ret = MESA_load_profile_string_nodef(profile, section, "log_path", log_path, sizeof(log_path));
	if(ret < 0){
		printf("MESA_prof_load: log_path not set, profile = %s, section = %s", profile, section);
		goto error_out;
	}
	ret = MESA_load_profile_int_nodef(profile, section, "log_level", &log_level);
	if(ret < 0){
		printf("MESA_prof_load: log_level not set, profile = %s, section = %s", profile, section);
		goto error_out;
	}
	local_logger = MESA_create_runtime_log_handle(log_path, log_level);
	if (unlikely(local_logger == NULL)){
		printf("Failed at create logger: %s", log_path);
		goto error_out;
	}
	g_kni_handle = ALLOC(struct kni_handle, 1);
    g_kni_handle->local_logger = local_logger;

	//kni_git_log
	KNI_LOG_ERROR(local_logger, "----------kni version = %s-----------", kni_git_verison);

	char deploy_mode[KNI_SYMBOL_MAX];
	ret = MESA_load_profile_string_def(profile, section, "deploy_mode", deploy_mode, sizeof(deploy_mode), "normal");
	if (strcmp(deploy_mode, "tap") == 0)
	{
		g_kni_handle->deploy_mode = KNI_DEPLOY_MODE_TAP;
	}
	else if (strcmp(deploy_mode, "normal") == 0)
	{
		g_kni_handle->deploy_mode = KNI_DEPLOY_MODE_NORMAL;
	}
	else
	{
		KNI_LOG_ERROR(local_logger, "KNI Invalid deploy_mode: %s\n", deploy_mode);
		goto error_out;
	}
	
	if(g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_NORMAL){
		ret = MESA_load_profile_int_nodef(profile, section, "tfe_node_count", &tfe_node_count);
		if(ret < 0){
			KNI_LOG_ERROR(local_logger, "MESA_prof_load: tfe_node_count not set, profile = %s, section = %s", profile, section);
			goto error_out;
		}
		if(tfe_node_count > TFE_COUNT_MAX){
			KNI_LOG_ERROR(local_logger, "tfe_node_count = %d, exceed the max_tfe_node_count %d", tfe_node_count, TFE_COUNT_MAX);
			goto error_out;
		}
		if(tfe_node_count <= 0){
			KNI_LOG_ERROR(local_logger, "tfe_node_count = %d, <= 0", tfe_node_count);
			goto error_out;
		}
	}
	ret = MESA_load_profile_string_nodef(profile, section, "manage_eth", manage_eth, sizeof(manage_eth));
	if(ret < 0){
		printf("MESA_prof_load: manage_eth not set, profile = %s, section = %s", profile, section);
		goto error_out;
	}
	char src_mac_addr_str[KNI_SYMBOL_MAX];
	char dst_mac_addr_str[KNI_SYMBOL_MAX];
	ret = MESA_load_profile_string_nodef(profile, section, "src_mac_addr", src_mac_addr_str, sizeof(src_mac_addr_str));
	if(ret < 0){
		KNI_LOG_ERROR(local_logger, "MESA_prof_load: src_mac_addr not set, profile = %s, section = %s", profile, section);
		goto error_out;
	}
	ret = MESA_load_profile_string_nodef(profile, section, "dst_mac_addr", dst_mac_addr_str, sizeof(dst_mac_addr_str));
	if(ret < 0){
		KNI_LOG_ERROR(local_logger, "MESA_prof_load: dst_mac_addr not set, profile = %s, section = %s", profile, section);
		goto error_out;
	}
	KNI_LOG_ERROR(local_logger, "MESA_prof_load, [%s]:\n log_path: %s\n log_level: %d\n tfe_node_count: %d\n manage_eth: %s\n deploy_mode: %s\n"
			"src_mac_addr: %s\n dst_mac_addr: %s", section, log_path, log_level, tfe_node_count, manage_eth, deploy_mode, src_mac_addr_str, dst_mac_addr_str);
	//ff:ee:dd:cc:bb:aa --->  0xff 0xee 0xdd 0xcc 0xbb 0xaa
	ret = sscanf(src_mac_addr_str, "%02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx",
    		&(g_kni_handle->src_mac_addr[0]), &(g_kni_handle->src_mac_addr[1]),
    		&(g_kni_handle->src_mac_addr[2]), &(g_kni_handle->src_mac_addr[3]),
    		&(g_kni_handle->src_mac_addr[4]), &(g_kni_handle->src_mac_addr[5]));
	if(ret != 6){
		KNI_LOG_ERROR(local_logger, "MESA_prof_load: src_mac_addr = invalid, ret = %d, profile = %s, section = %s", ret, profile, section);
		goto error_out;
	}
	ret = sscanf(dst_mac_addr_str, "%02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx",
    		&(g_kni_handle->dst_mac_addr[0]), &(g_kni_handle->dst_mac_addr[1]),
    		&(g_kni_handle->dst_mac_addr[2]), &(g_kni_handle->dst_mac_addr[3]),
    		&(g_kni_handle->dst_mac_addr[4]), &(g_kni_handle->dst_mac_addr[5]));
	if(ret != 6){
		KNI_LOG_ERROR(local_logger, "MESA_prof_load: dst_mac_addr = invalid, ret = %d, profile = %s, section = %s", ret, profile, section);
		goto error_out;
	}

	// get thread count
	g_kni_handle->thread_count = get_thread_count();
	if(g_kni_handle->thread_count <= 0){
		KNI_LOG_ERROR(local_logger, "Failed at get_thread_count, ret = %d");
		goto error_out;
	}

	//init marsio
	if(g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_NORMAL){
		g_kni_handle->marsio_handle = kni_marsio_init(profile, tfe_node_count);
		if(g_kni_handle->marsio_handle == NULL){
			KNI_LOG_ERROR(local_logger, "Failed at init marsio");
			goto error_out;
		}
	}
	else if (g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_TAP)
	{
		if (tfe_node_count != 1)
		{
			KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "under tap mode, only support one tfe node.");
			goto error_out;
		}

		if (MESA_load_profile_string_nodef(profile, "tap", "tap_name", tap_name, sizeof(tap_name)) < 0)
		{
			KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "under tap mode, tap_name is required.");
			goto error_out;
		}

		MESA_load_profile_int_nodef(profile, "io_uring", "enable_iouring", (int *)&g_kni_handle->iouring_conf.enable_iouring);
		MESA_load_profile_int_nodef(profile, "io_uring", "enable_debuglog", (int *)&g_kni_handle->iouring_conf.enable_debuglog);
		MESA_load_profile_int_nodef(profile, "io_uring", "ring_size", (int *)&g_kni_handle->iouring_conf.ring_size);
		MESA_load_profile_int_nodef(profile, "io_uring", "buff_size", (int *)&g_kni_handle->iouring_conf.buff_size);
		MESA_load_profile_int_nodef(profile, "io_uring", "flags", (int *)&g_kni_handle->iouring_conf.flags);
		MESA_load_profile_int_nodef(profile, "io_uring", "sq_thread_idle", (int *)&g_kni_handle->iouring_conf.sq_thread_idle);

		MESA_load_profile_int_nodef(profile, "tap", "tap_rps_enable", &g_kni_handle->tap_rps_enable);
		if (g_kni_handle->tap_rps_enable)
		{
			if (MESA_load_profile_string_nodef(profile, "tap", "tap_rps_mask", g_kni_handle->tap_rps_mask, sizeof(g_kni_handle->tap_rps_mask)) < 0)
			{
				KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "under tap mode, when enable tap_rps_enable, tap_rps_mask is required.");
				goto error_out;
			}
		}

		int tap_allow_mutilthread = 0;
		MESA_load_profile_int_nodef(profile, "tap", "tap_allow_mutilthread", &tap_allow_mutilthread);
		if (tap_allow_mutilthread)
		{
			uint32_t bpf_debug_log = 0;
			uint32_t bpf_hash_mode = 2;
			uint32_t bpf_queue_num = get_thread_count();
			char bpf_obj[1024] = {0};

			MESA_load_profile_int_nodef(profile, "tap", "bpf_debug_log", (int *)&bpf_debug_log);
			MESA_load_profile_int_nodef(profile, "tap", "bpf_hash_mode", (int *)&bpf_hash_mode);
			ret = MESA_load_profile_string_nodef(profile, "tap", "bpf_obj", bpf_obj, sizeof(bpf_obj));
			if (ret < 0)
			{
				KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "under tap mode, when enable tap_allow_mutilthread, bpf_obj is required.");
				goto error_out;
			}

			g_kni_handle->tap_bpf_ctx = kni_tap_global_load_rss_bpf(bpf_obj, bpf_queue_num, bpf_hash_mode, bpf_debug_log, local_logger);
			if (g_kni_handle->tap_bpf_ctx == NULL)
			{
				goto error_out;
			}
		}
		else
		{
			if (get_thread_count() > 1)
			{
				KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "under tap mode, when disable tap_allow_mutilthread, only support one work thread.");
				goto error_out;
			}
		}
	}


	//init_filedstat
	fs_handle = fs_init(profile);
	if(fs_handle == NULL){
		KNI_LOG_ERROR(local_logger, "Failed at init field_stat");
		goto error_out;
	}
    g_kni_fs_handle = fs_handle;

	n_pxy_fs_thread = get_thread_count() + 1;
	g_kni_handle->proxy_fieldstat = proxy_fieldstat_init(profile, "proxy_hits", 
														 n_pxy_fs_thread,
														 local_logger);
	if(g_kni_handle->proxy_fieldstat == NULL)
	{
		KNI_LOG_ERROR(local_logger, "Failed at init proxy fieldstat");
		goto error_out;
	}

	//init traceid2pme_htable
	struct kni_htable_opt opt;
	memset(&opt, 0, sizeof(opt));
	kni_get_htable_opt(&opt, profile, "traceid2pme_htable", NULL, (void*)traceid2pme_htable_expire_notify_cb, local_logger);
	traceid2pme_htable = kni_create_htable((char*)"traceid2pme_htable", &opt, local_logger);
	if(traceid2pme_htable == NULL){
		KNI_LOG_ERROR(local_logger, "Failed at create traceid2pme_htable");
		goto error_out;
	}
	g_kni_handle->traceid2pme_htable = traceid2pme_htable;

	//init tuple2stream_htable
	g_kni_handle->threads_handle = ALLOC(struct per_thread_handle, g_kni_handle->thread_count);
	memset(&opt, 0, sizeof(opt));
	kni_get_htable_opt(&opt, profile, "tuple2stream_htable", (void*)tuple2stream_htable_data_free_cb, NULL, local_logger);
	for(int i = 0; i < g_kni_handle->thread_count; i++){
		MESA_htable_handle tuple2stream_htable = kni_create_htable((char*)"tuple2stream_htable", &opt, local_logger);
		if(tuple2stream_htable == NULL){
			KNI_LOG_ERROR(local_logger, "Failed at kni_create_htable, table = tuple2stream_htable");
			goto error_out;
		}
		g_kni_handle->threads_handle[i].tuple2stream_htable = tuple2stream_htable;
		if (g_kni_handle->deploy_mode == KNI_DEPLOY_MODE_TAP)
		{
			g_kni_handle->threads_handle[i].tap_fd = kni_tap_open_per_thread(tap_name, IFF_TAP | IFF_NO_PI | IFF_MULTI_QUEUE, kni_tap_get_bpf_prog_fd(g_kni_handle->tap_bpf_ctx), local_logger);
			if (g_kni_handle->threads_handle[i].tap_fd == -1)
			{
				goto error_out;
			}
			if (g_kni_handle->iouring_conf.enable_iouring)
			{
#if (SUPPORT_LIBURING)
				g_kni_handle->threads_handle[i].iouring_handle = io_uring_handle_create(
					g_kni_handle->iouring_conf.ring_size, g_kni_handle->iouring_conf.buff_size,
					g_kni_handle->iouring_conf.flags, g_kni_handle->iouring_conf.sq_thread_idle);
				if (g_kni_handle->threads_handle[i].iouring_handle == NULL)
				{
					KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "Failed at kni create iouring");
					goto error_out;
				}

				io_uring_add_read(i);
#else
				KNI_LOG_ERROR(local_logger, TAP_RSS_LOG_TAG "Unsupport io_uring !!!");
				exit(-1);
#endif
			}
			else
			{
				g_kni_handle->threads_handle[i].buff_size = KNI_MTU;
				g_kni_handle->threads_handle[i].buff = ALLOC(char, g_kni_handle->threads_handle[i].buff_size);
			}
		}
	}

	if (g_kni_handle->tap_rps_enable)
	{
#if 0
		char rps_cmd[2048];
		for (int i = 0; i < g_kni_handle->thread_count; i++)
		{
			memset(rps_cmd, 0, sizeof(rps_cmd));
			snprintf(rps_cmd, sizeof(rps_cmd), "echo %s > /sys/class/net/%s/queues/rx-%d/rps_cpus", g_kni_handle->tap_rps_mask, tap_name, i);
			system(rps_cmd);
			KNI_LOG_DEBUG(local_logger, TAP_RSS_LOG_TAG "'%s'", rps_cmd);
		}
#endif
		if (set_rps(local_logger, tap_name, g_kni_handle->thread_count, g_kni_handle->tap_rps_mask) == -1)
		{
			goto error_out;
		}
	}

	//init ssl dynamic bypass htable
	ret = ssl_dynamic_bypass_htable_init(profile, local_logger);
	if(ret < 0){
		goto error_out;
	}

	//init dabloom_handle
    ret = dup_traffic_dabloom_init(profile, local_logger);
	if(ret < 0){
		goto error_out;
	}

	//init array last_tfe_dispatch_index and read security policy id for tsg-diagnose shunt
	MESA_load_profile_int_def(profile, "tsg_diagnose", "enabled", &g_kni_handle->tsg_diagnose_enable, 1);
	KNI_LOG_ERROR(local_logger, "tsg_diagnose: MESA_prof_load, tsg_diagnose:\n enabled: %d", g_kni_handle->tsg_diagnose_enable);
	g_kni_handle->arr_last_tfe_dispatch_index = ALLOC(int,g_kni_handle->thread_count);
	memset(&g_kni_handle->secpolicyid_shunt_tsg_diagnose, 0, sizeof(g_kni_handle->secpolicyid_shunt_tsg_diagnose));
	ret = MESA_load_profile_uint_range(profile, "tsg_diagnose", "security_policy_id", TSG_DIAGNOSE_POLICY_CNT, (unsigned int *)g_kni_handle->secpolicyid_shunt_tsg_diagnose.id_arr);
	g_kni_handle->secpolicyid_shunt_tsg_diagnose.id_num = ret;
	if(ret <= 0){
		KNI_LOG_ERROR(local_logger, "Fail get security_policy_id for tsg diagnose, tsg_diagnose no action to security policy id");
	}
	else{
		for(int i = 0; i < g_kni_handle->secpolicyid_shunt_tsg_diagnose.id_num; i++){
			if(g_kni_handle->secpolicyid_shunt_tsg_diagnose.id_arr[i] <= 0)
				KNI_LOG_ERROR(local_logger, "Tsg diagnose, security policy id  is not allowed to be equal to and to be lesser than 0");
			else{
				KNI_LOG_ERROR(local_logger, "tsg_diagnose: MESA_prof_load, tsg_diagnose: policy id:%d",g_kni_handle->secpolicyid_shunt_tsg_diagnose.id_arr[i]);
			}
			
		}
	}
	
	//init proxy tcp option maat 
	ret = pxy_tcp_option_rule_init(profile, local_logger);
	if(ret < 0){
		KNI_LOG_ERROR(local_logger, "Failed at init pxy_tcp_option_rule");
		goto error_out;
	}
/*
	//register customer for share session attribute
	MESA_load_profile_string_def(profile, "share_session_attribute", "SESSION_ATTRIBUTE_LABEL", label_buff, sizeof(label_buff), "TSG_MASTER_INTERNAL_LABEL");
	g_kni_handle->session_attribute_id = project_customer_register(label_buff, PROJECT_VAL_TYPE_STRUCT);
	if(g_kni_handle->session_attribute_id < 0)
	{
		KNI_LOG_ERROR(local_logger,"Register %s failed; please check :%s ",label_buff,profile);
		return -1;
	}
*/
	//init tfe_mgr
	_tfe_mgr = tfe_mgr_init(tfe_node_count, profile, g_kni_handle->deploy_mode, local_logger);
	if(_tfe_mgr == NULL){
		KNI_LOG_ERROR(local_logger, "Failed at init tfe_mgr");
		goto error_out;
	}
	g_kni_handle->_tfe_mgr = _tfe_mgr;

	//create thread_tfe_cmsg_receiver
	cmsg_receiver_args = ALLOC(struct thread_tfe_cmsg_receiver_args, 1);
	cmsg_receiver_args->logger = local_logger;
	strncpy(cmsg_receiver_args->profile, profile, strnlen(profile, sizeof(cmsg_receiver_args->profile) - 1));
	ret = pthread_create(&thread_id, NULL, thread_tfe_cmsg_receiver, (void *)cmsg_receiver_args);
	if(unlikely(ret != 0)){
		KNI_LOG_ERROR(local_logger, "Failed at pthread_create, thread_func = thread_tfe_cmsg_receiver, errno = %d, errmsg = %s", errno, strerror(errno));
		FREE(&cmsg_receiver_args);
		goto error_out;
	}
	return 0;

error_out:
    if(g_kni_handle->arr_last_tfe_dispatch_index)
		FREE(&(g_kni_handle->arr_last_tfe_dispatch_index));
	kni_destroy();
	exit(0);
}

#if (SUPPORT_LIBURING)

static int io_uring_add_write(int thread_seq, char *raw_data, uint16_t raw_len, addr_type_t addr_type)
{
	int tap_fd = g_kni_handle->threads_handle[thread_seq].tap_fd;
	struct io_uring_handle *handle = g_kni_handle->threads_handle[thread_seq].iouring_handle;
	struct io_uring *ring = &handle->ring;

	struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
	if (sqe == NULL)
	{
		KNI_LOG_ERROR(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "kni the io_uring submission queue is full, drop write packet");
		return -1;
	}

	struct user_data *conn = io_uring_user_data_create(tap_fd, EVTYPE_WRITE, handle->buff_size);
	conn->vec.iov_len = sizeof(struct ethhdr) + raw_len;
	add_ether_header(conn->vec.iov_base, raw_data, raw_len, addr_type);

	io_uring_prep_writev(sqe, conn->sockfd, &conn->vec, 1, 0);
	io_uring_sqe_set_data(sqe, conn);

	if (g_kni_handle->iouring_conf.enable_debuglog)
	{
		KNI_LOG_DEBUG(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "Add event: write, thread: %02d, sockfd: %02d, sqe: %p, user_data: %p, iov_base: %p, iovec_len: %04d",
					  thread_seq, conn->sockfd, sqe, sqe->user_data, conn->vec.iov_base, conn->vec.iov_len);
	}
	io_uring_submit(ring);

	return 0;
}

static int io_uring_add_read(int thread_seq)
{
	int tap_fd = g_kni_handle->threads_handle[thread_seq].tap_fd;
	struct io_uring_handle *handle = g_kni_handle->threads_handle[thread_seq].iouring_handle;
	struct io_uring *ring = &handle->ring;

	struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
	if (sqe == NULL)
	{
		KNI_LOG_ERROR(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "kni the io_uring submission queue is full, drop read packet");
		return -1;
	}

	struct user_data *conn = io_uring_user_data_create(tap_fd, EVTYPE_READ, handle->buff_size);
	io_uring_prep_readv(sqe, conn->sockfd, &conn->vec, 1, 0);
	io_uring_sqe_set_data(sqe, conn);

	if (g_kni_handle->iouring_conf.enable_debuglog)
	{
		KNI_LOG_DEBUG(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "Add event: read, thread: %02d, sockfd: %02d, sqe: %p, user_data: %p, iov_base: %p, iovec_len: %04d",
					  thread_seq, conn->sockfd, sqe, sqe->user_data, conn->vec.iov_base, conn->vec.iov_len);
	}
	io_uring_submit(ring);

	return 0;
}

static int io_uring_polling(int thread_seq)
{
	int tap_fd = g_kni_handle->threads_handle[thread_seq].tap_fd;
	MESA_htable_handle tuple2stream_htable = g_kni_handle->threads_handle[thread_seq].tuple2stream_htable;
	struct io_uring_handle *handle = g_kni_handle->threads_handle[thread_seq].iouring_handle;
	struct io_uring *ring = &handle->ring;

	struct io_uring_cqe *cqes[MAX_BATCH_CQE_NUM];
	int ret = io_uring_peek_batch_cqe(ring, cqes, MAX_BATCH_CQE_NUM);
	if (ret <= 0)
	{
		return -1;
	}

	for (int i = 0; i < ret; i++)
	{
		struct io_uring_cqe *cqe = cqes[i];
		if (cqe == NULL)
		{
			KNI_LOG_ERROR(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "kni iouring cqe return NULL");
			continue;
		}
		if (cqe->user_data == NULL)
		{
			KNI_LOG_ERROR(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "kni iouring cqe->user_data return NULL");
			continue;
		}

		struct user_data *conn = (struct user_data *)cqe->user_data;
		if (g_kni_handle->iouring_conf.enable_debuglog)
		{
			KNI_LOG_DEBUG(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "Handle event: %s, thread: %02d, sockfd: %02d, cqe: %p, user_data: %p, iov_base: %p, iovec_len: %04d, cqe->res: %04d",
						  (conn->type == EVTYPE_READ ? "read" : (conn->type == EVTYPE_WRITE ? "write" : "unknown")),
						  thread_seq, conn->sockfd, cqe, cqe->user_data, conn->vec.iov_base, conn->vec.iov_len, cqe->res);
		}
		switch (conn->type)
		{
		case EVTYPE_READ:
			if (cqe->res > 0)
			{
				struct ethhdr *ether_hdr = (struct ethhdr *)conn->vec.iov_base;
				tuple2stream_htable_search(tuple2stream_htable, ether_hdr, thread_seq);
			}
			else
			{
				KNI_LOG_ERROR(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "kni iouring tapfd[%d] read error, (%d, %s)", tap_fd, -cqe->res, strerror(-cqe->res));
			}
			io_uring_user_data_destory(conn);
			io_uring_add_read(thread_seq);
			break;
		case EVTYPE_WRITE:
			if (cqe->res < 0)
			{
				KNI_LOG_ERROR(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "kni iouring tapfd[%d] write error, (%d, %s)", tap_fd, -cqe->res, strerror(-cqe->res));
			}
			io_uring_user_data_destory(conn);
			break;
		case EVTYPE_UNKNOWN:
			assert(0);
			break;
		default:
			assert(0);
			break;
		}
		io_uring_cqe_seen(&handle->ring, cqe);
	}
	return 0;
}

#else

static int io_uring_add_write(int thread_seq, char *raw_data, uint16_t raw_len, addr_type_t addr_type)
{
	KNI_LOG_ERROR(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "Unsupport io_uring !!!");
	exit(-1);

	return -1;
}

static int io_uring_add_read(int thread_seq)
{
	KNI_LOG_ERROR(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "Unsupport io_uring !!!");
	exit(-1);

	return -1;
}

static int io_uring_polling(int thread_seq)
{
	KNI_LOG_ERROR(g_kni_handle->local_logger, TAP_RSS_LOG_TAG "Unsupport io_uring !!!");
	exit(-1);

	return -1;
}

#endif