stellar-stellar/infra/session_manager/session_manager_rte.c

#include <assert.h>

#include "packet_helper.h"
#include "packet_filter.h"
#include "session_pool.h"
#include "session_table.h"
#include "session_timer.h"
#include "session_filter.h"
#include "session_internal.h"
#include "session_transition.h"
#include "session_manager_log.h"
#include "session_manager_cfg.h"
#include "session_manager_rte.h"

struct snowflake
{
    uint64_t seed;
    uint64_t sequence;
};

struct session_manager_rte
{
    struct session_queue evc_list;
    struct session_pool *sess_pool;
    struct session_table *tcp_table;
    struct session_table *udp_table;
    struct session_timer *sess_timer;

    struct packet_filter *dup_pkt_filter;
    struct session_filter *evc_sess_filter;

    struct session_manager_cfg cfg;
    struct session_manager_stat stat;

    // only used for session_set_discard() or session_manager_rte_record_duplicated_packet(), because the function is called by plugin and has no time input.
    uint64_t now_ms;
    uint64_t last_clean_expired_sess_ts;
    struct snowflake *sf;
};

/******************************************************************************
 * snowflake
 ******************************************************************************/

static struct snowflake *snowflake_new(uint64_t seed)
{
    struct snowflake *sf = (struct snowflake *)calloc(1, sizeof(struct snowflake));
    if (sf == NULL)
    {
        return NULL;
    }

    sf->seed = seed & 0xFFFFF;
    sf->sequence = 0;

    return sf;
}

static void snowflake_free(struct snowflake *sf)
{
    if (sf != NULL)
    {
        free(sf);
        sf = NULL;
    }
}

static uint64_t snowflake_generate(struct snowflake *sf, uint64_t now_sec)
{
    /*
     * high -> low
     * +------+------------------+----------------+------------------------+---------------------------+
     * | 1bit |  12bit device_id | 8bit thread_id | 28bit timestamp in sec | 15bit sequence per thread |
     * +------+------------------+----------------+------------------------+---------------------------+
     */
#define MAX_ID_PER_THREAD (32768)
#define MAX_ID_BASE_TIME (268435456L)
    uint64_t id = 0;
    uint64_t id_per_thread = (sf->sequence++) % MAX_ID_PER_THREAD;
    uint64_t id_base_time = now_sec % MAX_ID_BASE_TIME;

    id = (sf->seed << 43) | (id_base_time << 15) | (id_per_thread);

    return id;
}

/******************************************************************************
 * address range
 ******************************************************************************/

static int ipv4_in_range(const struct in_addr *addr, const struct in_addr *start, const struct in_addr *end)
{
    return (memcmp(addr, start, sizeof(struct in_addr)) >= 0 && memcmp(addr, end, sizeof(struct in_addr)) <= 0);
}

static int ipv6_in_range(const struct in6_addr *addr, const struct in6_addr *start, const struct in6_addr *end)
{
    return (memcmp(addr, start, sizeof(struct in6_addr)) >= 0 && memcmp(addr, end, sizeof(struct in6_addr)) <= 0);
}

/******************************************************************************
 * TCP reassembly
 ******************************************************************************/

static void tcp_clean(struct session_manager_rte *sess_mgr_rte, struct session *sess)
{
    struct tcp_reassembly *c2s_tcp_reass = sess->tcp_halfs[FLOW_TYPE_C2S].tcp_reass;
    struct tcp_reassembly *s2c_tcp_reass = sess->tcp_halfs[FLOW_TYPE_S2C].tcp_reass;
    struct tcp_segment *seg;
    if (c2s_tcp_reass)
    {
        while ((seg = tcp_reassembly_expire(c2s_tcp_reass, UINT64_MAX)))
        {
            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);
            sess_mgr_rte->stat.tcp_segs_freed++;
            tcp_segment_free(seg);
        }
        tcp_reassembly_free(c2s_tcp_reass);
    }
    if (s2c_tcp_reass)
    {
        while ((seg = tcp_reassembly_expire(s2c_tcp_reass, UINT64_MAX)))
        {
            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);
            sess_mgr_rte->stat.tcp_segs_freed++;
            tcp_segment_free(seg);
        }
        tcp_reassembly_free(s2c_tcp_reass);
    }
}

static int tcp_init(struct session_manager_rte *sess_mgr_rte, struct session *sess)
{
    if (!sess_mgr_rte->cfg.tcp_reassembly.enable)
    {
        return 0;
    }

    sess->tcp_halfs[FLOW_TYPE_C2S].tcp_reass = tcp_reassembly_new(sess_mgr_rte->cfg.tcp_reassembly.timeout_ms, sess_mgr_rte->cfg.tcp_reassembly.buffered_segments_max);
    sess->tcp_halfs[FLOW_TYPE_S2C].tcp_reass = tcp_reassembly_new(sess_mgr_rte->cfg.tcp_reassembly.timeout_ms, sess_mgr_rte->cfg.tcp_reassembly.buffered_segments_max);
    if (sess->tcp_halfs[FLOW_TYPE_C2S].tcp_reass == NULL || sess->tcp_halfs[FLOW_TYPE_S2C].tcp_reass == NULL)
    {
        tcp_clean(sess_mgr_rte, sess);
        return -1;
    }

    SESSION_MANAGER_LOG_DEBUG("session %lu %s new c2s tcp tcp_reass %p, s2c tcp tcp_reass %p", session_get_id(sess), session_get_readable_addr(sess),
                              sess->tcp_halfs[FLOW_TYPE_C2S].tcp_reass, sess->tcp_halfs[FLOW_TYPE_S2C].tcp_reass);
    return 0;
}

static void tcp_update(struct session_manager_rte *sess_mgr_rte, struct session *sess, enum flow_type type, const struct layer_internal *tcp_layer)
{
    struct tcp_segment *seg;
    struct tcphdr *hdr = (struct tcphdr *)tcp_layer->hdr_ptr;
    struct tcp_half *half = &sess->tcp_halfs[type];
    uint8_t flags = tcp_hdr_get_flags(hdr);
    uint16_t len = tcp_layer->pld_len;

    if ((flags & TH_SYN) && half->isn == 0)
    {
        half->isn = tcp_hdr_get_seq(hdr);
    }
    half->flags = flags;
    half->history |= flags;
    half->seq = tcp_hdr_get_seq(hdr);
    half->ack = tcp_hdr_get_ack(hdr);
    half->len = tcp_layer->pld_len;

    if (!sess_mgr_rte->cfg.tcp_reassembly.enable)
    {
        if (len)
        {
            session_inc_stat(sess, type, STAT_TCP_SEGMENTS_RECEIVED, 1);
            session_inc_stat(sess, type, STAT_TCP_PAYLOADS_RECEIVED, len);
            sess_mgr_rte->stat.tcp_segs_input++;

            session_inc_stat(sess, type, STAT_TCP_SEGMENTS_INORDER, 1);
            session_inc_stat(sess, type, STAT_TCP_PAYLOADS_INORDER, len);
            sess_mgr_rte->stat.tcp_segs_inorder++;

            half->inorder_seg.data = tcp_layer->pld_ptr;
            half->inorder_seg.len = len;
            half->inorder_seg_consumed = 0;
        }
        return;
    }

    if (unlikely(flags & TH_SYN))
    {
        // len > 0 is SYN with data (TCP Fast Open)
        tcp_reassembly_set_recv_next(half->tcp_reass, len ? half->seq : half->seq + 1);

        if (unlikely(flags & TH_ACK))
        {
            // current packet is SYN-ACK (S2C), if C2S has not received SYN-ACK, set C2S recv_next
            struct tcp_half *peer = &sess->tcp_halfs[FLOW_TYPE_C2S];
            if (type == FLOW_TYPE_S2C && peer->history == 0 && tcp_reassembly_get_recv_next(peer->tcp_reass) == 0)
            {
                tcp_reassembly_set_recv_next(peer->tcp_reass, half->ack);
            }
        }
    }

    seg = tcp_reassembly_expire(half->tcp_reass, sess_mgr_rte->now_ms);
    if (seg)
    {
        session_inc_stat(sess, type, STAT_TCP_SEGMENTS_EXPIRED, 1);
        session_inc_stat(sess, type, STAT_TCP_PAYLOADS_EXPIRED, seg->len);
        sess_mgr_rte->stat.tcp_segs_timeout++;

        session_inc_stat(sess, type, STAT_TCP_SEGMENTS_RELEASED, 1);
        session_inc_stat(sess, type, STAT_TCP_PAYLOADS_RELEASED, seg->len);
        sess_mgr_rte->stat.tcp_segs_freed++;

        tcp_segment_free(seg);
    }

    if (len)
    {
        session_inc_stat(sess, type, STAT_TCP_SEGMENTS_RECEIVED, 1);
        session_inc_stat(sess, type, STAT_TCP_PAYLOADS_RECEIVED, len);
        sess_mgr_rte->stat.tcp_segs_input++;

        uint32_t rcv_nxt = tcp_reassembly_get_recv_next(half->tcp_reass);
        // in order
        if (half->seq == rcv_nxt)
        {
            session_inc_stat(sess, type, STAT_TCP_SEGMENTS_INORDER, 1);
            session_inc_stat(sess, type, STAT_TCP_PAYLOADS_INORDER, len);
            sess_mgr_rte->stat.tcp_segs_inorder++;

            half->inorder_seg.data = tcp_layer->pld_ptr;
            half->inorder_seg.len = len;
            half->inorder_seg_consumed = 0;
            tcp_reassembly_inc_recv_next(half->tcp_reass, len);
        }
        // retransmission
        else if (uint32_before(uint32_add(half->seq, len), rcv_nxt))
        {
            session_inc_stat(sess, type, STAT_TCP_SEGMENTS_RETRANSMIT, 1);
            session_inc_stat(sess, type, STAT_TCP_PAYLOADS_RETRANSMIT, len);
            sess_mgr_rte->stat.tcp_segs_retransmited++;
        }
        else if ((seg = tcp_segment_new(half->seq, tcp_layer->pld_ptr, len)))
        {
            switch (tcp_reassembly_push(half->tcp_reass, seg, sess_mgr_rte->now_ms))
            {
            case -2:
                session_inc_stat(sess, type, STAT_TCP_SEGMENTS_RETRANSMIT, 1);
                session_inc_stat(sess, type, STAT_TCP_PAYLOADS_RETRANSMIT, len);
                sess_mgr_rte->stat.tcp_segs_retransmited++;
                tcp_segment_free(seg);
                break;
            case -1:
                session_inc_stat(sess, type, STAT_TCP_SEGMENTS_NOSPACE, 1);
                session_inc_stat(sess, type, STAT_TCP_PAYLOADS_NOSPACE, len);
                sess_mgr_rte->stat.tcp_segs_omitted_too_many++;
                tcp_segment_free(seg);
                break;
            case 0:
                session_inc_stat(sess, type, STAT_TCP_SEGMENTS_BUFFERED, 1);
                session_inc_stat(sess, type, STAT_TCP_PAYLOADS_BUFFERED, len);
                sess_mgr_rte->stat.tcp_segs_buffered++;
                break;
            case 1:
                session_inc_stat(sess, type, STAT_TCP_SEGMENTS_OVERLAP, 1);
                session_inc_stat(sess, type, STAT_TCP_PAYLOADS_OVERLAP, len);
                sess_mgr_rte->stat.tcp_segs_overlapped++;

                session_inc_stat(sess, type, STAT_TCP_SEGMENTS_BUFFERED, 1);
                session_inc_stat(sess, type, STAT_TCP_PAYLOADS_BUFFERED, len);
                sess_mgr_rte->stat.tcp_segs_buffered++;
                break;
            default:
                assert(0);
                break;
            }
        }
        else
        {
            session_inc_stat(sess, type, STAT_TCP_SEGMENTS_NOSPACE, 1);
            session_inc_stat(sess, type, STAT_TCP_PAYLOADS_NOSPACE, len);
            sess_mgr_rte->stat.tcp_segs_omitted_too_many++;
        }
    }
}

/******************************************************************************
 * flow type
 ******************************************************************************/

// TODO
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_TYPE_C2S;
    }
    else if (src_port < dst_port)
    {
        return FLOW_TYPE_S2C;
    }
    else
    {
        // if port is equal, first packet is C2S
        return FLOW_TYPE_C2S;
    }
}

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_TYPE_C2S;
    }
    else
    {
        return FLOW_TYPE_S2C;
    }
}

/******************************************************************************
 * session manager rte -- session update
 ******************************************************************************/

static void session_update(struct session_manager_rte *sess_mgr_rte, 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(sess_mgr_rte->sf, sess_mgr_rte->now_ms / 1000);
        session_set_id(sess, sess_id);
        enum packet_direction pkt_dir = packet_get_direction(pkt);
        if (type == FLOW_TYPE_C2S)
        {
            session_set_tuple6(sess, key);
            if (pkt_dir == PACKET_DIRECTION_OUTGOING) // Internal -> External
            {
                session_set_direction(sess, SESSION_DIRECTION_OUTBOUND);
            }
            else
            {
                session_set_direction(sess, SESSION_DIRECTION_INBOUND);
            }
            tuple6_to_str(key, sess->tuple_str, sizeof(sess->tuple_str));
        }
        else
        {
            struct tuple6 out;
            tuple6_reverse(key, &out);
            session_set_tuple6(sess, &out);
            if (pkt_dir == PACKET_DIRECTION_OUTGOING) // Internal -> External
            {
                session_set_direction(sess, SESSION_DIRECTION_INBOUND);
            }
            else
            {
                session_set_direction(sess, SESSION_DIRECTION_OUTBOUND);
            }
            tuple6_to_str(&out, sess->tuple_str, sizeof(sess->tuple_str));
        }

        session_set_timestamp(sess, SESSION_TIMESTAMP_START, sess_mgr_rte->now_ms);
        switch (key->ip_proto)
        {
        case IPPROTO_TCP:
            session_set_type(sess, SESSION_TYPE_TCP);
            break;
        case IPPROTO_UDP:
            session_set_type(sess, SESSION_TYPE_UDP);
            break;
        default:
            assert(0);
            break;
        }
    }

    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, type))
    {
        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_flow_type(sess, type);
    session_set_timestamp(sess, SESSION_TIMESTAMP_LAST, sess_mgr_rte->now_ms);
    session_set_current_state(sess, next_state);
}

/******************************************************************************
 * session manager rte -- bypass packet
 ******************************************************************************/

static int session_manager_rte_bypass_packet_on_tcp_table_limit(struct session_manager_rte *sess_mgr_rte, const struct tuple6 *key)
{
    if (key->ip_proto == IPPROTO_TCP && sess_mgr_rte->stat.tcp_sess_used >= sess_mgr_rte->cfg.tcp_session_max)
    {
        sess_mgr_rte->stat.tcp_pkts_bypass_table_full++;
        return 1;
    }
    return 0;
}

static int session_manager_rte_bypass_packet_on_udp_table_limit(struct session_manager_rte *sess_mgr_rte, const struct tuple6 *key)
{
    if (key->ip_proto == IPPROTO_UDP && sess_mgr_rte->stat.udp_sess_used >= sess_mgr_rte->cfg.udp_session_max)
    {
        sess_mgr_rte->stat.udp_pkts_bypass_table_full++;
        return 1;
    }
    return 0;
}

static int session_manager_rte_bypass_packet_on_session_evicted(struct session_manager_rte *sess_mgr_rte, const struct tuple6 *key)
{
    if (sess_mgr_rte->cfg.evicted_session_bloom_filter.enable && session_filter_lookup(sess_mgr_rte->evc_sess_filter, key, sess_mgr_rte->now_ms))
    {
        sess_mgr_rte->stat.udp_pkts_bypass_session_evicted++;
        return 1;
    }

    return 0;
}

static int session_manager_rte_bypass_duplicated_packet(struct session_manager_rte *sess_mgr_rte, struct session *sess, const struct packet *pkt, const struct tuple6 *key)
{
    if (sess_mgr_rte->cfg.duplicated_packet_bloom_filter.enable == 0)
    {
        return 0;
    }

    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(sess_mgr_rte->dup_pkt_filter, pkt, sess_mgr_rte->now_ms))
        {
            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:
                sess_mgr_rte->stat.tcp_pkts_bypass_duplicated++;
                break;
            case SESSION_TYPE_UDP:
                sess_mgr_rte->stat.udp_pkts_bypass_duplicated++;
                break;
            default:
                assert(0);
                break;
            }
            session_set_duplicate_traffic(sess);

            session_set_current_packet(sess, pkt);
            session_set_flow_type(sess, type);
            return 1;
        }
        else
        {
            packet_filter_add(sess_mgr_rte->dup_pkt_filter, pkt, sess_mgr_rte->now_ms);
            return 0;
        }
    }

    return 0;
}

/******************************************************************************
 * session manager rte -- lookup/new/update/evicte session
 ******************************************************************************/

static void session_manager_rte_evicte_session(struct session_manager_rte *sess_mgr_rte, struct session *sess, int reason)
{
    if (sess == NULL)
    {
        return;
    }

    // when session add to evicted queue, session lifetime is over
    enum session_state curr_state = session_get_current_state(sess);
    enum session_state next_state = session_transition_run(curr_state, reason);
    session_transition_log(sess, curr_state, next_state, reason);
    session_set_current_state(sess, next_state);
    if (!session_get_closing_reason(sess))
    {
        if (reason == PORT_REUSE_EVICT)
        {
            session_set_closing_reason(sess, CLOSING_BY_PORT_REUSE_EVICTED);
        }
        if (reason == LRU_EVICT)
        {
            session_set_closing_reason(sess, CLOSING_BY_LRU_EVICTED);
        }
    }
    session_timer_del(sess_mgr_rte->sess_timer, sess);
    TAILQ_INSERT_TAIL(&sess_mgr_rte->evc_list, sess, evc_tqe);

    switch (session_get_type(sess))
    {
    case SESSION_TYPE_TCP:
        SESSION_MANAGER_LOG_DEBUG("evicte tcp old session: %lu", session_get_id(sess));
        session_table_del(sess_mgr_rte->tcp_table, sess);
        SESS_MGR_STAT_UPDATE(&sess_mgr_rte->stat, curr_state, next_state, tcp);
        sess_mgr_rte->stat.tcp_sess_evicted++;
        break;
    case SESSION_TYPE_UDP:
        SESSION_MANAGER_LOG_DEBUG("evicte udp old session: %lu", session_get_id(sess));
        session_table_del(sess_mgr_rte->udp_table, sess);
        if (sess_mgr_rte->cfg.evicted_session_bloom_filter.enable)
        {
            session_filter_add(sess_mgr_rte->evc_sess_filter, session_get_tuple6(sess), sess_mgr_rte->now_ms);
        }
        SESS_MGR_STAT_UPDATE(&sess_mgr_rte->stat, curr_state, next_state, udp);
        sess_mgr_rte->stat.udp_sess_evicted++;
        break;
    default:
        assert(0);
        break;
    }
}

static struct session *session_manager_rte_lookup_tcp_session(struct session_manager_rte *sess_mgr_rte, const struct packet *pkt, const struct tuple6 *key)
{
    struct session *sess = session_table_find_tuple6(sess_mgr_rte->tcp_table, key, 0);
    if (sess == NULL)
    {
        return NULL;
    }

    const struct layer_internal *tcp_layer = packet_get_innermost_layer(pkt, LAYER_PROTO_TCP);
    const struct tcphdr *hdr = (const struct tcphdr *)tcp_layer->hdr_ptr;
    uint8_t flags = tcp_hdr_get_flags(hdr);
    if ((flags & TH_SYN) == 0)
    {
        return sess;
    }

    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
    {
        // TCP port reuse, evict old session
        session_manager_rte_evicte_session(sess_mgr_rte, sess, PORT_REUSE_EVICT);
        return NULL;
    }
    else
    {
        // TCP SYN retransmission
        return sess;
    }
}

static struct session *session_manager_rte_lookup_udp_session(struct session_manager_rte *sess_mgr_rte, const struct packet *pkt __attribute__((unused)), const struct tuple6 *key)
{
    return session_table_find_tuple6(sess_mgr_rte->udp_table, key, 0);
}

static struct session *session_manager_rte_new_tcp_session(struct session_manager_rte *sess_mgr_rte, const struct packet *pkt, const struct tuple6 *key)
{
    const struct layer_internal *tcp_layer = packet_get_innermost_layer(pkt, LAYER_PROTO_TCP);
    const struct tcphdr *hdr = (const struct tcphdr *)tcp_layer->hdr_ptr;
    uint8_t flags = tcp_hdr_get_flags(hdr);
    if (!(flags & TH_SYN))
    {
        sess_mgr_rte->stat.tcp_pkts_bypass_session_not_found++;
        return NULL;
    }

    // tcp table full evict old session
    if (sess_mgr_rte->cfg.evict_old_on_tcp_table_limit && sess_mgr_rte->stat.tcp_sess_used >= sess_mgr_rte->cfg.tcp_session_max - RX_BURST_MAX)
    {
        struct session *evic_sess = session_table_find_lru(sess_mgr_rte->tcp_table);
        session_manager_rte_evicte_session(sess_mgr_rte, evic_sess, LRU_EVICT);
    }

    enum flow_type type = (flags & TH_ACK) ? FLOW_TYPE_S2C : FLOW_TYPE_C2S;
    struct session *sess = session_pool_pop(sess_mgr_rte->sess_pool);
    if (sess == NULL)
    {
        assert(0);
        return NULL;
    }
    session_init(sess);
    sess->sess_mgr_rte = sess_mgr_rte;
    sess->sess_mgr_stat = &sess_mgr_rte->stat;

    enum session_state next_state = session_transition_run(SESSION_STATE_INIT, TCP_SYN);
    session_update(sess_mgr_rte, sess, next_state, pkt, key, type);
    session_transition_log(sess, SESSION_STATE_INIT, next_state, TCP_SYN);

    if (tcp_init(sess_mgr_rte, sess) == -1)
    {
        assert(0);
        session_pool_push(sess_mgr_rte->sess_pool, sess);
        return NULL;
    }
    tcp_update(sess_mgr_rte, sess, type, tcp_layer);

    uint64_t timeout = (flags & TH_ACK) ? sess_mgr_rte->cfg.tcp_timeout_ms.handshake : sess_mgr_rte->cfg.tcp_timeout_ms.init;
    session_timer_update(sess_mgr_rte->sess_timer, sess, sess_mgr_rte->now_ms + timeout);
    session_table_add(sess_mgr_rte->tcp_table, sess);

    if (sess_mgr_rte->cfg.duplicated_packet_bloom_filter.enable)
    {
        packet_filter_add(sess_mgr_rte->dup_pkt_filter, pkt, sess_mgr_rte->now_ms);
    }

    SESS_MGR_STAT_INC(&sess_mgr_rte->stat, next_state, tcp);
    sess_mgr_rte->stat.tcp_sess_used++;
    sess_mgr_rte->stat.history_tcp_sessions++;

    return sess;
}

static struct session *session_manager_rte_new_udp_session(struct session_manager_rte *sess_mgr_rte, const struct packet *pkt, const struct tuple6 *key)
{
    // udp table full evict old session
    if (sess_mgr_rte->cfg.evict_old_on_udp_table_limit && sess_mgr_rte->stat.udp_sess_used >= sess_mgr_rte->cfg.udp_session_max - RX_BURST_MAX)
    {
        struct session *evic_sess = session_table_find_lru(sess_mgr_rte->udp_table);
        session_manager_rte_evicte_session(sess_mgr_rte, evic_sess, LRU_EVICT);
    }

    struct session *sess = session_pool_pop(sess_mgr_rte->sess_pool);
    if (sess == NULL)
    {
        assert(sess);
        return NULL;
    }
    session_init(sess);
    sess->sess_mgr_rte = sess_mgr_rte;
    sess->sess_mgr_stat = &sess_mgr_rte->stat;

    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(sess_mgr_rte, sess, next_state, pkt, key, type);
    session_transition_log(sess, SESSION_STATE_INIT, next_state, UDP_DATA);

    session_timer_update(sess_mgr_rte->sess_timer, sess, sess_mgr_rte->now_ms + sess_mgr_rte->cfg.udp_timeout_ms.data);
    session_table_add(sess_mgr_rte->udp_table, sess);

    SESS_MGR_STAT_INC(&sess_mgr_rte->stat, next_state, udp);
    sess_mgr_rte->stat.udp_sess_used++;
    sess_mgr_rte->stat.history_udp_sessions++;

    return sess;
}

static int session_manager_rte_update_tcp_session(struct session_manager_rte *sess_mgr_rte, struct session *sess, const struct packet *pkt, const struct tuple6 *key)
{
    const struct layer_internal *tcp_layer = packet_get_innermost_layer(pkt, LAYER_PROTO_TCP);
    const struct tcphdr *hdr = (const struct tcphdr *)tcp_layer->hdr_ptr;
    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;
    inputs |= (flags & TH_FIN) ? TCP_FIN : NONE;
    inputs |= (flags & TH_RST) ? TCP_RST : NONE;
    inputs |= tcp_layer->pld_len ? TCP_DATA : NONE;

    // update state
    enum session_state curr_state = session_get_current_state(sess);
    enum session_state next_state = session_transition_run(curr_state, inputs);

    // update session
    session_update(sess_mgr_rte, sess, next_state, pkt, key, type);
    session_transition_log(sess, curr_state, next_state, inputs);

    // update tcp
    tcp_update(sess_mgr_rte, 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, (type == FLOW_TYPE_C2S ? CLOSING_BY_CLIENT_FIN : CLOSING_BY_SERVER_FIN));
        }
        if (flags & TH_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[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)
    {
    case SESSION_STATE_OPENING:
        if (flags & TH_SYN)
        {
            timeout = (flags & TH_ACK) ? sess_mgr_rte->cfg.tcp_timeout_ms.handshake : sess_mgr_rte->cfg.tcp_timeout_ms.init;
        }
        else
        {
            timeout = sess_mgr_rte->cfg.tcp_timeout_ms.data;
        }
        break;
    case SESSION_STATE_ACTIVE:
        timeout = sess_mgr_rte->cfg.tcp_timeout_ms.data;
        break;
    case SESSION_STATE_CLOSING:
        if (flags & TH_FIN)
        {
            timeout = (peer->history & TH_FIN) ? sess_mgr_rte->cfg.tcp_timeout_ms.time_wait : sess_mgr_rte->cfg.tcp_timeout_ms.half_closed;
        }
        else if (flags & TH_RST)
        {
            // if fin is received, the expected sequence number should be increased by 1
            uint32_t expected = (peer->history & TH_FIN) ? peer->ack + 1 : peer->ack;
            timeout = (expected == curr->seq) ? sess_mgr_rte->cfg.tcp_timeout_ms.time_wait : sess_mgr_rte->cfg.tcp_timeout_ms.unverified_rst;
        }
        else
        {
            timeout = sess_mgr_rte->cfg.tcp_timeout_ms.data;
        }
        break;
    case SESSION_STATE_DISCARD:
        timeout = sess_mgr_rte->cfg.tcp_timeout_ms.discard_default;
        break;
    default:
        assert(0);
        break;
    }
    session_timer_update(sess_mgr_rte->sess_timer, sess, sess_mgr_rte->now_ms + timeout);

    SESS_MGR_STAT_UPDATE(&sess_mgr_rte->stat, curr_state, next_state, tcp);

    return 0;
}

static int session_manager_rte_update_udp_session(struct session_manager_rte *sess_mgr_rte, struct session *sess, const struct packet *pkt, const struct tuple6 *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(sess_mgr_rte, 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)
    {
        session_timer_update(sess_mgr_rte->sess_timer, sess, sess_mgr_rte->now_ms + sess_mgr_rte->cfg.udp_timeout_ms.discard_default);
    }
    else
    {
        session_timer_update(sess_mgr_rte->sess_timer, sess, sess_mgr_rte->now_ms + sess_mgr_rte->cfg.udp_timeout_ms.data);
    }

    SESS_MGR_STAT_UPDATE(&sess_mgr_rte->stat, curr_state, next_state, udp);

    return 0;
}

/******************************************************************************
 * session manager rte -- public
 ******************************************************************************/

struct session_manager_rte *session_manager_rte_new(const struct session_manager_cfg *sess_mgr_cfg, uint64_t now_ms)
{
    struct session_manager_rte *sess_mgr_rte = (struct session_manager_rte *)calloc(1, sizeof(struct session_manager_rte));
    if (sess_mgr_rte == NULL)
    {
        return NULL;
    }
    memcpy(&sess_mgr_rte->cfg, sess_mgr_cfg, sizeof(struct session_manager_cfg));

    sess_mgr_rte->sess_pool = session_pool_new(sess_mgr_rte->cfg.tcp_session_max + sess_mgr_rte->cfg.udp_session_max);
    sess_mgr_rte->tcp_table = session_table_new();
    sess_mgr_rte->udp_table = session_table_new();
    sess_mgr_rte->sess_timer = session_timer_new(now_ms);
    if (sess_mgr_rte->sess_pool == NULL || sess_mgr_rte->tcp_table == NULL || sess_mgr_rte->udp_table == NULL || sess_mgr_rte->sess_timer == NULL)
    {
        goto error;
    }
    if (sess_mgr_rte->cfg.evicted_session_bloom_filter.enable)
    {
        sess_mgr_rte->evc_sess_filter = session_filter_new(sess_mgr_rte->cfg.evicted_session_bloom_filter.capacity,
                                                           sess_mgr_rte->cfg.evicted_session_bloom_filter.time_window_ms,
                                                           sess_mgr_rte->cfg.evicted_session_bloom_filter.error_rate, now_ms);
        if (sess_mgr_rte->evc_sess_filter == NULL)
        {
            goto error;
        }
    }
    if (sess_mgr_rte->cfg.duplicated_packet_bloom_filter.enable)
    {
        sess_mgr_rte->dup_pkt_filter = packet_filter_new(sess_mgr_rte->cfg.duplicated_packet_bloom_filter.capacity,
                                                         sess_mgr_rte->cfg.duplicated_packet_bloom_filter.time_window_ms,
                                                         sess_mgr_rte->cfg.duplicated_packet_bloom_filter.error_rate, now_ms);
        if (sess_mgr_rte->dup_pkt_filter == NULL)
        {
            goto error;
        }
    }
    sess_mgr_rte->sf = snowflake_new(sess_mgr_rte->cfg.session_id_seed);
    if (sess_mgr_rte->sf == NULL)
    {
        goto error;
    }

    TAILQ_INIT(&sess_mgr_rte->evc_list);
    session_transition_init();
    sess_mgr_rte->now_ms = now_ms;
    sess_mgr_rte->last_clean_expired_sess_ts = now_ms;

    return sess_mgr_rte;

error:
    session_manager_rte_free(sess_mgr_rte);
    return NULL;
}

void session_manager_rte_free(struct session_manager_rte *sess_mgr_rte)
{
    struct session *sess;
    if (sess_mgr_rte)
    {
        // free all evicted session
        while ((sess = TAILQ_FIRST(&sess_mgr_rte->evc_list)))
        {
            TAILQ_REMOVE(&sess_mgr_rte->evc_list, sess, evc_tqe);
            session_manager_rte_free_session(sess_mgr_rte, sess);
        }
        // free all udp session
        while (sess_mgr_rte->udp_table && (sess = session_table_find_lru(sess_mgr_rte->udp_table)))
        {
            session_manager_rte_free_session(sess_mgr_rte, sess);
        }
        // free all tcp session
        while (sess_mgr_rte->tcp_table && (sess = session_table_find_lru(sess_mgr_rte->tcp_table)))
        {
            session_manager_rte_free_session(sess_mgr_rte, sess);
        }
        if (sess_mgr_rte->cfg.evicted_session_bloom_filter.enable)
        {
            session_filter_free(sess_mgr_rte->evc_sess_filter);
        }
        if (sess_mgr_rte->cfg.duplicated_packet_bloom_filter.enable)
        {
            packet_filter_free(sess_mgr_rte->dup_pkt_filter);
        }
        snowflake_free(sess_mgr_rte->sf);
        session_timer_free(sess_mgr_rte->sess_timer);
        session_table_free(sess_mgr_rte->udp_table);
        session_table_free(sess_mgr_rte->tcp_table);
        session_pool_free(sess_mgr_rte->sess_pool);
        free(sess_mgr_rte);
        sess_mgr_rte = NULL;
    }
}

struct session *session_manager_rte_new_session(struct session_manager_rte *sess_mgr_rte, const struct packet *pkt, uint64_t now_ms)
{
    sess_mgr_rte->now_ms = now_ms;

    struct tuple6 key;
    if (packet_get_innermost_tuple6(pkt, &key))
    {
        return NULL;
    }
    switch (key.ip_proto)
    {
    case IPPROTO_TCP:
        if (session_manager_rte_bypass_packet_on_tcp_table_limit(sess_mgr_rte, &key))
        {
            return NULL;
        }
        return session_manager_rte_new_tcp_session(sess_mgr_rte, pkt, &key);
    case IPPROTO_UDP:
        if (session_manager_rte_bypass_packet_on_session_evicted(sess_mgr_rte, &key))
        {
            return NULL;
        }
        if (session_manager_rte_bypass_packet_on_udp_table_limit(sess_mgr_rte, &key))
        {
            return NULL;
        }
        return session_manager_rte_new_udp_session(sess_mgr_rte, pkt, &key);
    default:
        return NULL;
    }
}

void session_manager_rte_free_session(struct session_manager_rte *sess_mgr_rte, struct session *sess)
{
    if (sess)
    {
        SESSION_MANAGER_LOG_DEBUG("session %lu closed (%s)", session_get_id(sess), closing_reason_to_str(session_get_closing_reason(sess)));

        session_timer_del(sess_mgr_rte->sess_timer, sess);
        switch (session_get_type(sess))
        {
        case SESSION_TYPE_TCP:
            tcp_clean(sess_mgr_rte, sess);
            if (session_table_find_sessid(sess_mgr_rte->tcp_table, session_get_id(sess), 0) == sess)
            {
                session_table_del(sess_mgr_rte->tcp_table, sess);
            }
            SESS_MGR_STAT_DEC(&sess_mgr_rte->stat, session_get_current_state(sess), tcp);
            sess_mgr_rte->stat.tcp_sess_used--;
            break;
        case SESSION_TYPE_UDP:
            if (session_table_find_sessid(sess_mgr_rte->udp_table, session_get_id(sess), 0) == sess)
            {
                session_table_del(sess_mgr_rte->udp_table, sess);
            }
            SESS_MGR_STAT_DEC(&sess_mgr_rte->stat, session_get_current_state(sess), udp);
            sess_mgr_rte->stat.udp_sess_used--;
            break;
        default:
            assert(0);
            break;
        }

        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_flow_type(sess, FLOW_TYPE_NONE);
        session_init(sess);
        session_pool_push(sess_mgr_rte->sess_pool, sess);
        sess = NULL;
    }
}

struct session *session_manager_rte_lookup_session_by_packet(struct session_manager_rte *sess_mgr_rte, const struct packet *pkt)
{
    struct tuple6 key;
    if (packet_get_innermost_tuple6(pkt, &key))
    {
        return NULL;
    }
    switch (key.ip_proto)
    {
    case IPPROTO_UDP:
        return session_manager_rte_lookup_udp_session(sess_mgr_rte, pkt, &key);
    case IPPROTO_TCP:
        return session_manager_rte_lookup_tcp_session(sess_mgr_rte, pkt, &key);
    default:
        return NULL;
    }
}

struct session *session_manager_rte_lookup_session_by_id(struct session_manager_rte *sess_mgr_rte, uint64_t sess_id)
{
    struct session *sess = NULL;
    sess = session_table_find_sessid(sess_mgr_rte->tcp_table, sess_id, 1);
    if (sess)
    {
        return sess;
    }

    sess = session_table_find_sessid(sess_mgr_rte->udp_table, sess_id, 1);
    if (sess)
    {
        return sess;
    }

    return NULL;
}

int session_manager_rte_update_session(struct session_manager_rte *sess_mgr_rte, struct session *sess, const struct packet *pkt, uint64_t now_ms)
{
    sess_mgr_rte->now_ms = now_ms;

    struct tuple6 key;
    if (packet_get_innermost_tuple6(pkt, &key))
    {
        return -1;
    }
    if (session_manager_rte_bypass_duplicated_packet(sess_mgr_rte, sess, pkt, &key))
    {
        return -1;
    }
    switch (session_get_type(sess))
    {
    case SESSION_TYPE_TCP:
        return session_manager_rte_update_tcp_session(sess_mgr_rte, sess, pkt, &key);
    case SESSION_TYPE_UDP:
        return session_manager_rte_update_udp_session(sess_mgr_rte, sess, pkt, &key);
    default:
        return -1;
    }
}

struct session *session_manager_rte_get_expired_session(struct session_manager_rte *sess_mgr_rte, uint64_t now_ms)
{
    sess_mgr_rte->now_ms = now_ms;

    struct session *sess = session_timer_expire(sess_mgr_rte->sess_timer, now_ms);
    if (sess)
    {
        enum session_state curr_state = session_get_current_state(sess);
        enum session_state next_state = session_transition_run(curr_state, TIMEOUT);
        session_transition_log(sess, curr_state, next_state, TIMEOUT);
        session_set_current_state(sess, next_state);

        switch (session_get_type(sess))
        {
        case SESSION_TYPE_TCP:
            SESS_MGR_STAT_UPDATE(&sess_mgr_rte->stat, curr_state, next_state, tcp);
            break;
        case SESSION_TYPE_UDP:
            SESS_MGR_STAT_UPDATE(&sess_mgr_rte->stat, curr_state, next_state, udp);
            break;
        default:
            assert(0);
            break;
        }

        // next state is closed, need to free session
        if (next_state == SESSION_STATE_CLOSED)
        {
            if (!session_get_closing_reason(sess))
            {
                session_set_closing_reason(sess, CLOSING_BY_TIMEOUT);
            }
            return sess;
        }
        // next state is closing, only update timeout
        else
        {
            switch (session_get_type(sess))
            {
            case SESSION_TYPE_TCP:
                session_timer_update(sess_mgr_rte->sess_timer, sess, now_ms + sess_mgr_rte->cfg.tcp_timeout_ms.data);
                break;
            case SESSION_TYPE_UDP:
                session_timer_update(sess_mgr_rte->sess_timer, sess, now_ms + sess_mgr_rte->cfg.udp_timeout_ms.data);
                break;
            default:
                assert(0);
                break;
            }
            return NULL;
        }
    }

    return NULL;
}

struct session *session_manager_rte_get_evicted_session(struct session_manager_rte *sess_mgr_rte)
{
    struct session *sess = TAILQ_FIRST(&sess_mgr_rte->evc_list);
    if (sess)
    {
        TAILQ_REMOVE(&sess_mgr_rte->evc_list, sess, evc_tqe);
    }
    return sess;
}

uint64_t session_manager_rte_clean_session(struct session_manager_rte *sess_mgr_rte, uint64_t now_ms, struct session *cleaned_sess_ptr[], uint64_t array_size)
{
    sess_mgr_rte->now_ms = now_ms;
    struct session *sess = NULL;
    uint64_t cleaned_sess_num = 0;
    uint64_t expired_sess_num = 0;

    uint8_t expired_sess_canbe_clean = 0;
    if (now_ms - sess_mgr_rte->last_clean_expired_sess_ts >= sess_mgr_rte->cfg.expire_period_ms ||
        now_ms == UINT64_MAX)
    {
        expired_sess_canbe_clean = 1;
    }

    for (uint64_t i = 0; i < array_size; i++)
    {
        // frist clean evicted session
        sess = session_manager_rte_get_evicted_session(sess_mgr_rte);
        if (sess)
        {
            cleaned_sess_ptr[cleaned_sess_num++] = sess;
        }
        // then clean expired session
        else
        {
            if (expired_sess_canbe_clean && expired_sess_num < sess_mgr_rte->cfg.expire_batch_max)
            {
                sess_mgr_rte->last_clean_expired_sess_ts = now_ms;
                sess = session_manager_rte_get_expired_session(sess_mgr_rte, now_ms);
                if (sess)
                {
                    cleaned_sess_ptr[cleaned_sess_num++] = sess;
                    expired_sess_num++;
                }
                else
                {
                    break;
                }
            }
            else
            {
                break;
            }
        }
    }

    return cleaned_sess_num;
}

uint64_t session_manager_rte_scan_session(struct session_manager_rte *sess_mgr_rte, const struct session_scan_opts *opts, uint64_t mached_sess_id[], uint64_t array_size)
{
    uint64_t capacity = 0;
    uint64_t max_loop = 0;
    uint64_t mached_sess_num = 0;
    const struct session *sess = NULL;
    const struct tuple6 *tuple = NULL;

    if (sess_mgr_rte == NULL || opts == NULL || mached_sess_id == NULL || array_size == 0)
    {
        return mached_sess_num;
    }
    if (opts->count == 0)
    {
        return mached_sess_num;
    }
    capacity = sess_mgr_rte->cfg.tcp_session_max + sess_mgr_rte->cfg.udp_session_max;
    if (opts->cursor >= capacity)
    {
        return mached_sess_num;
    }

    max_loop = MIN(capacity, opts->cursor + opts->count);
    for (uint64_t i = opts->cursor; i < max_loop; i++)
    {
        sess = session_pool_get0(sess_mgr_rte->sess_pool, i);
        tuple = session_get_tuple6(sess);
        if (session_get_current_state(sess) == SESSION_STATE_INIT)
        {
            continue;
        }

        if ((opts->flags & SESSION_SCAN_TYPE) && opts->type != session_get_type(sess))
        {
            continue;
        }
        if ((opts->flags & SESSION_SCAN_STATE) && opts->state != session_get_current_state(sess))
        {
            continue;
        }
        if ((opts->flags & SESSION_SCAN_CREATE_TIME) &&
            (session_get_timestamp(sess, SESSION_TIMESTAMP_START) < opts->create_time_ms[0] ||
             session_get_timestamp(sess, SESSION_TIMESTAMP_START) > opts->create_time_ms[1]))
        {
            continue;
        }
        if ((opts->flags & SESSION_SCAN_LASPKT_TIME) &&
            (session_get_timestamp(sess, SESSION_TIMESTAMP_LAST) < opts->laspkt_time_ms[0] ||
             session_get_timestamp(sess, SESSION_TIMESTAMP_LAST) > opts->laspkt_time_ms[1]))
        {
            continue;
        }
        if ((opts->flags & SESSION_SCAN_SPORT) && opts->src_port != tuple->src_port)
        {
            continue;
        }
        if ((opts->flags & SESSION_SCAN_DPORT) && opts->dst_port != tuple->dst_port)
        {
            continue;
        }
        if (opts->flags & SESSION_SCAN_SIP)
        {
            if (opts->addr_family != tuple->addr_family)
            {
                continue;
            }
            if ((opts->addr_family == AF_INET) && !ipv4_in_range(&tuple->src_addr.v4, &opts->src_addr[0].v4, &opts->src_addr[1].v4))
            {
                continue;
            }
            if ((opts->addr_family == AF_INET6) && !ipv6_in_range(&tuple->src_addr.v6, &opts->src_addr[0].v6, &opts->src_addr[1].v6))
            {
                continue;
            }
        }
        if (opts->flags & SESSION_SCAN_DIP)
        {
            if (opts->addr_family != tuple->addr_family)
            {
                continue;
            }
            if ((opts->addr_family == AF_INET) && !ipv4_in_range(&tuple->dst_addr.v4, &opts->dst_addr[0].v4, &opts->dst_addr[1].v4))
            {
                continue;
            }
            if ((opts->addr_family == AF_INET6) && !ipv6_in_range(&tuple->dst_addr.v6, &opts->dst_addr[0].v6, &opts->dst_addr[1].v6))
            {
                continue;
            }
        }

        mached_sess_id[mached_sess_num++] = session_get_id(sess);
        if (mached_sess_num >= array_size)
        {
            break;
        }
    }

    SESSION_MANAGER_LOG_DEBUG("session scan => cursor: %lu, count: %lu, mached_sess_num: %lu", opts->cursor, opts->count, mached_sess_num);
    return mached_sess_num;
}

void session_manager_rte_record_duplicated_packet(struct session_manager_rte *sess_mgr_rte, const struct packet *pkt)
{
    if (sess_mgr_rte->cfg.duplicated_packet_bloom_filter.enable)
    {
        packet_filter_add(sess_mgr_rte->dup_pkt_filter, pkt, sess_mgr_rte->now_ms);
    }
}

struct session_manager_stat *session_manager_rte_get_stat(struct session_manager_rte *sess_mgr_rte)
{
    return &sess_mgr_rte->stat;
}

void session_set_discard(struct session *sess)
{
    struct session_manager_rte *sess_mgr_rte = sess->sess_mgr_rte;
    enum session_type type = session_get_type(sess);
    enum session_state curr_state = session_get_current_state(sess);
    enum session_state next_state = session_transition_run(curr_state, USER_CLOSE);
    session_transition_log(sess, curr_state, next_state, USER_CLOSE);
    session_set_current_state(sess, next_state);

    switch (type)
    {
    case SESSION_TYPE_TCP:
        session_timer_update(sess_mgr_rte->sess_timer, sess, sess_mgr_rte->now_ms + sess_mgr_rte->cfg.tcp_timeout_ms.discard_default);
        SESS_MGR_STAT_UPDATE(&sess_mgr_rte->stat, curr_state, next_state, tcp);
        break;
    case SESSION_TYPE_UDP:
        session_timer_update(sess_mgr_rte->sess_timer, sess, sess_mgr_rte->now_ms + sess_mgr_rte->cfg.udp_timeout_ms.discard_default);
        SESS_MGR_STAT_UPDATE(&sess_mgr_rte->stat, curr_state, next_state, udp);
        break;
    default:
        assert(0);
        break;
    }
}