gfwleak/stellar-stellar
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rename flow_direction to flow_type

Browse Source
This commit is contained in:
luwenpeng
2024-09-02 17:49:33 +08:00
parent f8ec4dc5a7
commit a8206cffc0
28 changed files with 598 additions and 599 deletions
Download Patch File Download Diff File Expand all files Collapse all files

170
infra/session_manager/session_manager.c
View File

@@ -165,15 +165,15 @@ static uint64_t snowflake_generate(struct snowflake *sf, uint64_t now_sec)
static void tcp_clean(struct session_manager *mgr, struct session *sess)
{
struct tcp_reassembly *c2s_ssembler = sess->tcp_halfs[FLOW_DIRECTION_C2S].assembler;
struct tcp_reassembly *s2c_ssembler = sess->tcp_halfs[FLOW_DIRECTION_S2C].assembler;
struct tcp_reassembly *c2s_ssembler = sess->tcp_halfs[FLOW_TYPE_C2S].assembler;
struct tcp_reassembly *s2c_ssembler = sess->tcp_halfs[FLOW_TYPE_S2C].assembler;
struct tcp_segment *seg;
if (c2s_ssembler)
{
while ((seg = tcp_reassembly_expire(c2s_ssembler, UINT64_MAX)))
{
session_inc_stat(sess, FLOW_DIRECTION_C2S, STAT_TCP_SEGMENTS_RELEASED, 1);
session_inc_stat(sess, FLOW_DIRECTION_C2S, STAT_TCP_PAYLOADS_RELEASED, seg->len);
session_inc_stat(sess, FLOW_TYPE_C2S, STAT_TCP_SEGMENTS_RELEASED, 1);
session_inc_stat(sess, FLOW_TYPE_C2S, STAT_TCP_PAYLOADS_RELEASED, seg->len);
mgr->stat.tcp_segs_freed++;
tcp_segment_free(seg);
}
@@ -183,8 +183,8 @@ static void tcp_clean(struct session_manager *mgr, struct session *sess)
{
while ((seg = tcp_reassembly_expire(s2c_ssembler, UINT64_MAX)))
{
session_inc_stat(sess, FLOW_DIRECTION_S2C, STAT_TCP_SEGMENTS_RELEASED, 1);
session_inc_stat(sess, FLOW_DIRECTION_S2C, STAT_TCP_PAYLOADS_RELEASED, seg->len);
session_inc_stat(sess, FLOW_TYPE_S2C, STAT_TCP_SEGMENTS_RELEASED, 1);
session_inc_stat(sess, FLOW_TYPE_S2C, STAT_TCP_PAYLOADS_RELEASED, seg->len);
mgr->stat.tcp_segs_freed++;
tcp_segment_free(seg);
}
@@ -199,9 +199,9 @@ static int tcp_init(struct session_manager *mgr, struct session *sess)
return 0;
}
sess->tcp_halfs[FLOW_DIRECTION_C2S].assembler = tcp_reassembly_new(mgr->cfg.tcp_reassembly.timeout_ms, mgr->cfg.tcp_reassembly.buffered_segments_max);
sess->tcp_halfs[FLOW_DIRECTION_S2C].assembler = tcp_reassembly_new(mgr->cfg.tcp_reassembly.timeout_ms, mgr->cfg.tcp_reassembly.buffered_segments_max);
if (sess->tcp_halfs[FLOW_DIRECTION_C2S].assembler == NULL || sess->tcp_halfs[FLOW_DIRECTION_S2C].assembler == NULL)
sess->tcp_halfs[FLOW_TYPE_C2S].assembler = tcp_reassembly_new(mgr->cfg.tcp_reassembly.timeout_ms, mgr->cfg.tcp_reassembly.buffered_segments_max);
sess->tcp_halfs[FLOW_TYPE_S2C].assembler = tcp_reassembly_new(mgr->cfg.tcp_reassembly.timeout_ms, mgr->cfg.tcp_reassembly.buffered_segments_max);
if (sess->tcp_halfs[FLOW_TYPE_C2S].assembler == NULL || sess->tcp_halfs[FLOW_TYPE_S2C].assembler == NULL)
{
tcp_clean(mgr, sess);
return -1;
@@ -209,17 +209,17 @@ static int tcp_init(struct session_manager *mgr, struct session *sess)
SESSION_LOG_DEBUG("session %lu %s new c2s tcp assembler %p, s2c tcp assembler %p",
session_get_id(sess), session_get0_readable_addr(sess),
sess->tcp_halfs[FLOW_DIRECTION_C2S].assembler,
sess->tcp_halfs[FLOW_DIRECTION_S2C].assembler);
sess->tcp_halfs[FLOW_TYPE_C2S].assembler,
sess->tcp_halfs[FLOW_TYPE_S2C].assembler);
return 0;
}
static void tcp_update(struct session_manager *mgr, struct session *sess, enum flow_direction dir, const struct layer_private *tcp_layer)
static void tcp_update(struct session_manager *mgr, struct session *sess, enum flow_type type, const struct layer_private *tcp_layer)
{
struct tcp_segment *seg;
struct tcphdr *hdr = (struct tcphdr *)tcp_layer->hdr_ptr;
struct tcp_half *half = &sess->tcp_halfs[dir];
struct tcp_half *half = &sess->tcp_halfs[type];
uint8_t flags = tcp_hdr_get_flags(hdr);
uint16_t len = tcp_layer->pld_len;
@@ -237,12 +237,12 @@ static void tcp_update(struct session_manager *mgr, struct session *sess, enum f
{
if (len)
{
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_RECEIVED, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_RECEIVED, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_RECEIVED, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_RECEIVED, len);
mgr->stat.tcp_segs_input++;
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_INORDER, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_INORDER, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_INORDER, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_INORDER, len);
mgr->stat.tcp_segs_inorder++;
half->in_order.data = tcp_layer->pld_ptr;
@@ -261,12 +261,12 @@ static void tcp_update(struct session_manager *mgr, struct session *sess, enum f
seg = tcp_reassembly_expire(half->assembler, mgr->now_ms);
if (seg)
{
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_EXPIRED, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_EXPIRED, seg->len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_EXPIRED, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_EXPIRED, seg->len);
mgr->stat.tcp_segs_timeout++;
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_RELEASED, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_RELEASED, seg->len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_RELEASED, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_RELEASED, seg->len);
mgr->stat.tcp_segs_freed++;
tcp_segment_free(seg);
@@ -274,16 +274,16 @@ static void tcp_update(struct session_manager *mgr, struct session *sess, enum f
if (len)
{
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_RECEIVED, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_RECEIVED, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_RECEIVED, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_RECEIVED, len);
mgr->stat.tcp_segs_input++;
uint32_t rcv_nxt = tcp_reassembly_get_recv_next(half->assembler);
// in order
if (half->seq == rcv_nxt)
{
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_INORDER, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_INORDER, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_INORDER, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_INORDER, len);
mgr->stat.tcp_segs_inorder++;
half->in_order.data = tcp_layer->pld_ptr;
@@ -294,8 +294,8 @@ static void tcp_update(struct session_manager *mgr, struct session *sess, enum f
// retransmission
else if (uint32_before(uint32_add(half->seq, len), rcv_nxt))
{
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_RETRANSMIT, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_RETRANSMIT, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_RETRANSMIT, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_RETRANSMIT, len);
mgr->stat.tcp_segs_retransmited++;
}
else if ((seg = tcp_segment_new(half->seq, tcp_layer->pld_ptr, len)))
@@ -303,29 +303,29 @@ static void tcp_update(struct session_manager *mgr, struct session *sess, enum f
switch (tcp_reassembly_push(half->assembler, seg, mgr->now_ms))
{
case -2:
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_RETRANSMIT, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_RETRANSMIT, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_RETRANSMIT, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_RETRANSMIT, len);
mgr->stat.tcp_segs_retransmited++;
tcp_segment_free(seg);
break;
case -1:
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_NOSPACE, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_NOSPACE, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_NOSPACE, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_NOSPACE, len);
mgr->stat.tcp_segs_omitted_too_many++;
tcp_segment_free(seg);
break;
case 0:
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_BUFFERED, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_BUFFERED, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_BUFFERED, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_BUFFERED, len);
mgr->stat.tcp_segs_buffered++;
break;
case 1:
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_OVERLAP, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_OVERLAP, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_OVERLAP, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_OVERLAP, len);
mgr->stat.tcp_segs_overlapped++;
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_BUFFERED, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_BUFFERED, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_BUFFERED, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_BUFFERED, len);
mgr->stat.tcp_segs_buffered++;
break;
default:
@@ -335,8 +335,8 @@ static void tcp_update(struct session_manager *mgr, struct session *sess, enum f
}
else
{
session_inc_stat(sess, dir, STAT_TCP_SEGMENTS_NOSPACE, 1);
session_inc_stat(sess, dir, STAT_TCP_PAYLOADS_NOSPACE, len);
session_inc_stat(sess, type, STAT_TCP_SEGMENTS_NOSPACE, 1);
session_inc_stat(sess, type, STAT_TCP_PAYLOADS_NOSPACE, len);
mgr->stat.tcp_segs_omitted_too_many++;
}
}
@@ -346,33 +346,33 @@ static void tcp_update(struct session_manager *mgr, struct session *sess, enum f
* session direction identify
******************************************************************************/
static enum flow_direction identify_direction_by_port(uint16_t src_port, uint16_t dst_port)
static enum flow_type identify_flow_type_by_port(uint16_t src_port, uint16_t dst_port)
{
// big port is client
if (src_port > dst_port)
{
return FLOW_DIRECTION_C2S;
return FLOW_TYPE_C2S;
}
else if (src_port < dst_port)
{
return FLOW_DIRECTION_S2C;
return FLOW_TYPE_S2C;
}
else
{
// if port is equal, first packet is C2S
return FLOW_DIRECTION_C2S;
return FLOW_TYPE_C2S;
}
}
static enum flow_direction identify_direction_by_history(const struct session *sess, const struct tuple6 *key)
static enum flow_type identify_flow_type_by_history(const struct session *sess, const struct tuple6 *key)
{
if (tuple6_cmp(session_get_tuple6(sess), key) == 0)
{
return FLOW_DIRECTION_C2S;
return FLOW_TYPE_C2S;
}
else
{
return FLOW_DIRECTION_S2C;
return FLOW_TYPE_S2C;
}
}
@@ -420,13 +420,13 @@ static int duplicated_packet_bypass(struct session_manager *mgr, struct session
return 0;
}
enum flow_direction dir = identify_direction_by_history(sess, key);
if (session_get_stat(sess, dir, STAT_RAW_PACKETS_RECEIVED) < 3 || session_has_duplicate_traffic(sess))
enum flow_type type = identify_flow_type_by_history(sess, key);
if (session_get_stat(sess, type, STAT_RAW_PACKETS_RECEIVED) < 3 || session_has_duplicate_traffic(sess))
{
if (packet_filter_lookup(mgr->dup_pkt_filter, pkt, mgr->now_ms))
{
session_inc_stat(sess, dir, STAT_DUPLICATE_PACKETS_BYPASS, 1);
session_inc_stat(sess, dir, STAT_DUPLICATE_BYTES_BYPASS, packet_get_raw_len(pkt));
session_inc_stat(sess, type, STAT_DUPLICATE_PACKETS_BYPASS, 1);
session_inc_stat(sess, type, STAT_DUPLICATE_BYTES_BYPASS, packet_get_raw_len(pkt));
switch (session_get_type(sess))
{
case SESSION_TYPE_TCP:
@@ -442,7 +442,7 @@ static int duplicated_packet_bypass(struct session_manager *mgr, struct session
session_set_duplicate_traffic(sess);
session_set_current_packet(sess, pkt);
session_set_current_flow_direction(sess, dir);
session_set_flow_type(sess, type);
return 1;
}
else
@@ -459,14 +459,14 @@ static int duplicated_packet_bypass(struct session_manager *mgr, struct session
* session manager utils
******************************************************************************/
static void session_update(struct session_manager *mgr, struct session *sess, enum session_state next_state, const struct packet *pkt, const struct tuple6 *key, enum flow_direction dir)
static void session_update(struct session_manager *mgr, struct session *sess, enum session_state next_state, const struct packet *pkt, const struct tuple6 *key, enum flow_type type)
{
if (session_get_current_state(sess) == SESSION_STATE_INIT)
{
uint64_t sess_id = snowflake_generate(mgr->sf, mgr->now_ms / 1000);
session_set_id(sess, sess_id);
enum packet_direction pkt_dir = packet_get_direction(pkt);
if (dir == FLOW_DIRECTION_C2S)
if (type == FLOW_TYPE_C2S)
{
session_set_tuple6(sess, key);
if (pkt_dir == PACKET_DIRECTION_OUTGOING) // Internal -> External
@@ -510,18 +510,18 @@ static void session_update(struct session_manager *mgr, struct session *sess, en
}
}
session_inc_stat(sess, dir, STAT_RAW_PACKETS_RECEIVED, 1);
session_inc_stat(sess, dir, STAT_RAW_BYTES_RECEIVED, packet_get_raw_len(pkt));
session_inc_stat(sess, type, STAT_RAW_PACKETS_RECEIVED, 1);
session_inc_stat(sess, type, STAT_RAW_BYTES_RECEIVED, packet_get_raw_len(pkt));
if (!session_get_first_packet(sess, dir))
if (!session_get_first_packet(sess, type))
{
session_set_first_packet(sess, dir, packet_dup(pkt));
session_set_route_ctx(sess, dir, packet_get_route_ctx(pkt));
session_set_sids(sess, dir, packet_get_sids(pkt));
session_set_first_packet(sess, type, packet_dup(pkt));
session_set_route_ctx(sess, type, packet_get_route_ctx(pkt));
session_set_sids(sess, type, packet_get_sids(pkt));
}
session_set_current_packet(sess, pkt);
session_set_current_flow_direction(sess, dir);
session_set_flow_type(sess, type);
session_set_timestamp(sess, SESSION_TIMESTAMP_LAST, mgr->now_ms);
session_set_current_state(sess, next_state);
}
@@ -592,8 +592,8 @@ static struct session *session_manager_lookup_tcp_session(struct session_manager
return sess;
}
enum flow_direction dir = identify_direction_by_history(sess, key);
struct tcp_half *half = &sess->tcp_halfs[dir];
enum flow_type type = identify_flow_type_by_history(sess, key);
struct tcp_half *half = &sess->tcp_halfs[type];
if ((half->isn && half->isn != tcp_hdr_get_seq(hdr)) || // recv SYN with different ISN
((half->history & TH_FIN) || (half->history & TH_RST))) // recv SYN after FIN or RST
{
@@ -626,7 +626,7 @@ static struct session *session_manager_new_tcp_session(struct session_manager *m
session_manager_evicte_session(mgr, evic_sess, LRU_EVICT);
}
enum flow_direction dir = (flags & TH_ACK) ? FLOW_DIRECTION_S2C : FLOW_DIRECTION_C2S;
enum flow_type type = (flags & TH_ACK) ? FLOW_TYPE_S2C : FLOW_TYPE_C2S;
struct session *sess = session_pool_pop(mgr->sess_pool);
if (sess == NULL)
{
@@ -638,7 +638,7 @@ static struct session *session_manager_new_tcp_session(struct session_manager *m
sess->mgr_stat = &mgr->stat;
enum session_state next_state = session_transition_run(SESSION_STATE_INIT, TCP_SYN);
session_update(mgr, sess, next_state, pkt, key, dir);
session_update(mgr, sess, next_state, pkt, key, type);
session_transition_log(sess, SESSION_STATE_INIT, next_state, TCP_SYN);
if (tcp_init(mgr, sess) == -1)
@@ -647,7 +647,7 @@ static struct session *session_manager_new_tcp_session(struct session_manager *m
session_pool_push(mgr->sess_pool, sess);
return NULL;
}
tcp_update(mgr, sess, dir, tcp_layer);
tcp_update(mgr, sess, type, tcp_layer);
uint64_t timeout = (flags & TH_ACK) ? mgr->cfg.tcp_timeout_ms.handshake : mgr->cfg.tcp_timeout_ms.init;
session_timer_update(mgr->sess_timer, sess, mgr->now_ms + timeout);
@@ -684,9 +684,9 @@ static struct session *session_manager_new_udp_session(struct session_manager *m
sess->mgr = mgr;
sess->mgr_stat = &mgr->stat;
enum flow_direction dir = identify_direction_by_port(ntohs(key->src_port), ntohs(key->dst_port));
enum flow_type type = identify_flow_type_by_port(ntohs(key->src_port), ntohs(key->dst_port));
enum session_state next_state = session_transition_run(SESSION_STATE_INIT, UDP_DATA);
session_update(mgr, sess, next_state, pkt, key, dir);
session_update(mgr, sess, next_state, pkt, key, type);
session_transition_log(sess, SESSION_STATE_INIT, next_state, UDP_DATA);
session_timer_update(mgr->sess_timer, sess, mgr->now_ms + mgr->cfg.udp_timeout_ms.data);
@@ -703,7 +703,7 @@ static int session_manager_update_tcp_session(struct session_manager *mgr, struc
{
const struct layer_private *tcp_layer = packet_get_innermost_layer(pkt, LAYER_PROTO_TCP);
const struct tcphdr *hdr = (const struct tcphdr *)tcp_layer->hdr_ptr;
enum flow_direction dir = identify_direction_by_history(sess, key);
enum flow_type type = identify_flow_type_by_history(sess, key);
uint8_t flags = tcp_hdr_get_flags(hdr);
int inputs = 0;
inputs |= (flags & TH_SYN) ? TCP_SYN : NONE;
@@ -716,28 +716,28 @@ static int session_manager_update_tcp_session(struct session_manager *mgr, struc
enum session_state next_state = session_transition_run(curr_state, inputs);
// update session
session_update(mgr, sess, next_state, pkt, key, dir);
session_update(mgr, sess, next_state, pkt, key, type);
session_transition_log(sess, curr_state, next_state, inputs);
// update tcp
tcp_update(mgr, sess, dir, tcp_layer);
tcp_update(mgr, sess, type, tcp_layer);
// set closing reason
if (next_state == SESSION_STATE_CLOSING && !session_get_closing_reason(sess))
{
if (flags & TH_FIN)
{
session_set_closing_reason(sess, (dir == FLOW_DIRECTION_C2S ? CLOSING_BY_CLIENT_FIN : CLOSING_BY_SERVER_FIN));
session_set_closing_reason(sess, (type == FLOW_TYPE_C2S ? CLOSING_BY_CLIENT_FIN : CLOSING_BY_SERVER_FIN));
}
if (flags & TH_RST)
{
session_set_closing_reason(sess, (dir == FLOW_DIRECTION_C2S ? CLOSING_BY_CLIENT_RST : CLOSING_BY_SERVER_RST));
session_set_closing_reason(sess, (type == FLOW_TYPE_C2S ? CLOSING_BY_CLIENT_RST : CLOSING_BY_SERVER_RST));
}
}
// update timeout
struct tcp_half *curr = &sess->tcp_halfs[dir];
struct tcp_half *peer = &sess->tcp_halfs[(dir == FLOW_DIRECTION_C2S ? FLOW_DIRECTION_S2C : FLOW_DIRECTION_C2S)];
struct tcp_half *curr = &sess->tcp_halfs[type];
struct tcp_half *peer = &sess->tcp_halfs[(type == FLOW_TYPE_C2S ? FLOW_TYPE_S2C : FLOW_TYPE_C2S)];
uint64_t timeout = 0;
switch (next_state)
{
@@ -786,10 +786,10 @@ static int session_manager_update_tcp_session(struct session_manager *mgr, struc
static int session_manager_update_udp_session(struct session_manager *mgr, struct session *sess, const struct packet *pkt, const struct tuple6 *key)
{
enum flow_direction dir = identify_direction_by_history(sess, key);
enum flow_type type = identify_flow_type_by_history(sess, key);
enum session_state curr_state = session_get_current_state(sess);
enum session_state next_state = session_transition_run(curr_state, UDP_DATA);
session_update(mgr, sess, next_state, pkt, key, dir);
session_update(mgr, sess, next_state, pkt, key, type);
session_transition_log(sess, curr_state, next_state, UDP_DATA);
if (session_get_current_state(sess) == SESSION_STATE_DISCARD)
@@ -1100,17 +1100,17 @@ void session_manager_free_session(struct session_manager *mgr, struct session *s
break;
}
packet_free((struct packet *)session_get_first_packet(sess, FLOW_DIRECTION_C2S));
packet_free((struct packet *)session_get_first_packet(sess, FLOW_DIRECTION_S2C));
session_set_first_packet(sess, FLOW_DIRECTION_C2S, NULL);
session_set_first_packet(sess, FLOW_DIRECTION_S2C, NULL);
session_clear_route_ctx(sess, FLOW_DIRECTION_C2S);
session_clear_route_ctx(sess, FLOW_DIRECTION_S2C);
session_clear_sids(sess, FLOW_DIRECTION_C2S);
session_clear_sids(sess, FLOW_DIRECTION_S2C);
packet_free((struct packet *)session_get_first_packet(sess, FLOW_TYPE_C2S));
packet_free((struct packet *)session_get_first_packet(sess, FLOW_TYPE_S2C));
session_set_first_packet(sess, FLOW_TYPE_C2S, NULL);
session_set_first_packet(sess, FLOW_TYPE_S2C, NULL);
session_clear_route_ctx(sess, FLOW_TYPE_C2S);
session_clear_route_ctx(sess, FLOW_TYPE_S2C);
session_clear_sids(sess, FLOW_TYPE_C2S);
session_clear_sids(sess, FLOW_TYPE_S2C);
session_set_current_state(sess, SESSION_STATE_INIT);
session_set_current_packet(sess, NULL);
session_set_current_flow_direction(sess, FLOW_DIRECTION_NONE);
session_set_flow_type(sess, FLOW_TYPE_NONE);
session_pool_push(mgr->sess_pool, sess);
sess = NULL;
}