stellar-stellar/src/session/session_manager.cpp

#include <stdlib.h>
#include <assert.h>

#include "timestamp.h"
#include "session_private.h"
#include "session_pool.h"
#include "session_table.h"
#include "session_timer.h"
#include "session_queue.h"
#include "session_manager.h"
#include "tcp_helpers.h"
#include "udp_helpers.h"
#include "packet_helpers.h"
#include "dupkt_filter.h"
#include "eviction_filter.h"

struct session_manager
{
     struct session_pool *sess_pool;
     struct session_table *tcp_sess_table;
     struct session_table *udp_sess_table;
     struct session_timer *sess_timer;
     struct session_queue *sess_evicted_queue;
     struct session_queue *sess_toclosed_queue;

     struct dupkt_filter *tcp_dupkt_filter;
     struct eviction_filter *udp_eviction_filter;

     struct session_manager_config config;

     /***************************************************************
      * session manager status
      ***************************************************************/

     // session number
     uint64_t tcp_sess_num;
     uint64_t tcp_opening_sess_num;
     uint64_t tcp_active_sess_num;
     uint64_t tcp_closing_sess_num;

     uint64_t udp_sess_num;
     uint64_t udp_opening_sess_num;
     uint64_t udp_active_sess_num;
     uint64_t udp_closing_sess_num;

     // packet filter status
     uint64_t npkts_miss_l4_proto; // fast forward

     uint64_t npkts_hit_tcp_evicted; // fast forward (miss session)
     uint64_t npkts_hit_tcp_dupkt;   // fast forward
     uint64_t npkts_hit_tcp_discard; // drop
     uint64_t npkts_hit_tcp_closing; // fast forward

     uint64_t npkts_hit_udp_evicted; // fast forward
};

/******************************************************************************
 * utils
 ******************************************************************************/

// OK
// return  0: success
// return -1: invalid config
static int session_manager_check_config(struct session_manager_config *config)
{
     if (config == NULL)
     {
          SESSION_LOG_ERROR("invalid config");
          return -1;
     }

     // max session number
     if (config->max_tcp_session_num < 2)
     {
          SESSION_LOG_ERROR("invalid max tcp session number");
          return -1;
     }
     if (config->max_udp_session_num < 2)
     {
          SESSION_LOG_ERROR("invalid max udp session number");
          return -1;
     }

     // TCP timeout config
     if (config->tcp_timeout_init < 1 || config->tcp_timeout_init > 60)
     {
          SESSION_LOG_ERROR("invalid tcp timeout init, support range: 1-60");
          return -1;
     }
     if (config->tcp_timeout_handshake < 1 || config->tcp_timeout_handshake > 60)
     {
          SESSION_LOG_ERROR("invalid tcp timeout handshake, support range: 1-60");
          return -1;
     }
     if (config->tcp_timeout_data < 1 || config->tcp_timeout_data > 15999999)
     {
          SESSION_LOG_ERROR("invalid tcp timeout data, support range: 1-15,999,999");
          return -1;
     }
     if (config->tcp_timeout_half_closed < 1 || config->tcp_timeout_half_closed > 604800)
     {
          SESSION_LOG_ERROR("invalid tcp timeout half closed, support range: 1-604,800");
          return -1;
     }
     if (config->tcp_timeout_discard < 1 || config->tcp_timeout_discard > 15999999)
     {
          SESSION_LOG_ERROR("invalid tcp timeout discard, support range: 1-15,999,999");
          return -1;
     }

     // UDP timeout config
     if (config->udp_timeout_data < 1 || config->udp_timeout_data > 15999999)
     {
          SESSION_LOG_ERROR("invalid udp timeout data, support range: 1-15,999,999");
          return -1;
     }

     // TCP duplicate packet filter config
     if (config->tcp_dupkt_filter_enable != 0 && config->tcp_dupkt_filter_enable != 1)
     {
          SESSION_LOG_ERROR("invalid tcp dupkt filter enable, support range: 0-1");
          return -1;
     }
     if (config->tcp_dupkt_filter_enable)
     {
          if (config->tcp_dupkt_filter_capacity == 0)
          {
               SESSION_LOG_ERROR("invalid tcp dupkt filter capacity");
               return -1;
          }
          if (config->tcp_dupkt_filter_timeout < 1 || config->tcp_dupkt_filter_timeout > 60)
          {
               SESSION_LOG_ERROR("invalid tcp dupkt filter timeout, support range: 1-60");
               return -1;
          }
          if (config->tcp_dupkt_filter_error_rate < 0 || config->tcp_dupkt_filter_error_rate > 1)
          {
               SESSION_LOG_ERROR("invalid tcp dupkt filter error rate, support range: 0-1");
               return -1;
          }
     }

     // UDP eviction filter config
     if (config->udp_eviction_filter_enable != 0 && config->udp_eviction_filter_enable != 1)
     {
          SESSION_LOG_ERROR("invalid udp eviction filter enable, support range: 0-1");
          return -1;
     }
     if (config->udp_eviction_filter_enable)
     {
          if (config->udp_eviction_filter_capacity == 0)
          {
               SESSION_LOG_ERROR("invalid udp eviction filter capacity");
               return -1;
          }
          if (config->udp_eviction_filter_timeout < 1 || config->udp_eviction_filter_timeout > 60)
          {
               SESSION_LOG_ERROR("invalid udp eviction filter timeout, support range: 1-60");
               return -1;
          }
          if (config->udp_eviction_filter_error_rate < 0 || config->udp_eviction_filter_error_rate > 1)
          {
               SESSION_LOG_ERROR("invalid udp eviction filter error rate, support range: 0-1");
               return -1;
          }
     }

     return 0;
}

// TODO
static uint64_t session_manager_alloc_session_id(void)
{
     return 0;
}

// OK
// return 1: duplicate packet
// return 0: not duplicate packet
static int session_manager_filter_session_dupkt(struct session_manager *mgr, struct session *sess, const struct packet *pkt, enum session_dir curr_dir)
{
     if (curr_dir == SESSION_DIR_C2S)
     {
          if (session_get_c2s_packets(sess) < 3)
          {
               goto dupkt_fitler;
          }
     }
     else if (curr_dir == SESSION_DIR_S2C)
     {
          if (session_get_s2c_packets(sess) < 3)
          {
               goto dupkt_fitler;
          }
     }

     if (session_get_dup_traffic_flag(sess) == DUP_TRAFFIC_YES)
     {
          goto dupkt_fitler;
     }
     else
     {
          return 0;
     }

dupkt_fitler:
     if (dupkt_filter_lookup(mgr->tcp_dupkt_filter, pkt))
     {
          return 1;
     }
     else
     {
          dupkt_filter_add(mgr->tcp_dupkt_filter, pkt);
          return 0;
     }
}

/******************************************************************************
 * built-in ex data
 ******************************************************************************/

// OK
static void session_free_packet_exdata(struct session *sess, uint8_t idx, void *ex_ptr, void *arg)
{
     if (ex_ptr)
     {
          packet_free((struct packet *)ex_ptr);
     }
}

// OK
static void session_free_pktmeta_exdata(struct session *sess, uint8_t idx, void *ex_ptr, void *arg)
{
     if (ex_ptr)
     {
          metadata_free((struct metadata *)ex_ptr);
     }
}

// OK
static void session_update_packet_exdata(struct session *sess, const struct packet *pkt, enum session_dir curr_dir)
{
     if (curr_dir == SESSION_DIR_C2S)
     {
          if (session_get0_ex_data(sess, c2s_1st_pkt_ex) == NULL)
          {
               session_set_ex_data(sess, c2s_1st_pkt_ex, packet_dup(pkt));
          }
     }
     else if (curr_dir == SESSION_DIR_S2C)
     {
          if (session_get0_ex_data(sess, s2c_1st_pkt_ex) == NULL)
          {
               session_set_ex_data(sess, s2c_1st_pkt_ex, packet_dup(pkt));
          }
     }
}

// OK
static void session_update_pktmeta_exdata(struct session *sess, const struct metadata *md, enum session_dir curr_dir)
{
     if (curr_dir == SESSION_DIR_C2S)
     {
          if (session_get0_ex_data(sess, c2s_1st_md_ex) == NULL)
          {
               session_set_ex_data(sess, c2s_1st_md_ex, metadata_dup(md));
          }
     }
     else if (curr_dir == SESSION_DIR_S2C)
     {
          if (session_get0_ex_data(sess, s2c_1st_md_ex) == NULL)
          {
               session_set_ex_data(sess, s2c_1st_md_ex, metadata_dup(md));
          }
     }
}

// OK
static void session_update_tcp_exdata(struct session *sess, const struct layer_record *tcp_layer, enum session_dir curr_dir)
{
     const struct tcphdr *hdr = (const struct tcphdr *)tcp_layer->hdr_ptr;
     uint64_t state = (uint64_t)session_get0_ex_data(sess, tcp_builtin_ex);
     if (tcp_hdr_has_flag_syn(hdr))
     {
          state |= (tcp_hdr_has_flag_ack(hdr) ? TCP_SYNACK_RECVED : TCP_SYN_RECVED);
     }
     if (tcp_hdr_has_flag_fin(hdr))
     {
          if (curr_dir == SESSION_DIR_C2S)
          {
               state |= TCP_C2S_FIN_RECVED;
          }
          else if (curr_dir == SESSION_DIR_S2C)
          {
               state |= TCP_S2C_FIN_RECVED;
          }
     }
     if (tcp_hdr_has_flag_rst(hdr))
     {
          if (curr_dir == SESSION_DIR_C2S)
          {
               state |= TCP_C2S_RST_RECVED;
          }
          else if (curr_dir == SESSION_DIR_S2C)
          {
               state |= TCP_S2C_RST_RECVED;
          }
     }
     if (tcp_layer->pld_len > 0)
     {
          if (curr_dir == SESSION_DIR_C2S)
          {
               state |= TCP_C2S_DATA_RECVED;
          }
          else if (curr_dir == SESSION_DIR_S2C)
          {
               state |= TCP_S2C_DATA_RECVED;
          }
     }

     session_set_ex_data(sess, tcp_builtin_ex, (void *)(state));
}

// OK
static void session_update_udp_exdata(struct session *sess, const struct layer_record *udp_layer, enum session_dir curr_dir)
{
     const struct udphdr *hdr = (const struct udphdr *)udp_layer->hdr_ptr;
     uint64_t state = (uint64_t)session_get0_ex_data(sess, udp_builtin_ex);
     if (curr_dir == SESSION_DIR_C2S)
     {
          state |= UDP_C2S_RECVED;
     }
     else if (curr_dir == SESSION_DIR_S2C)
     {
          state |= UDP_S2C_RECVED;
     }
     session_set_ex_data(sess, udp_builtin_ex, (void *)(state));
}

/******************************************************************************
 * judge session direction
 ******************************************************************************/

// OK
static enum session_dir judge_direction_by_tuple6(const struct tuple6 *key)
{
     uint16_t src_port = ntohs(key->src_port);
     uint16_t dst_port = ntohs(key->dst_port);

     // big port is client
     if (src_port > dst_port)
     {
          return SESSION_DIR_C2S;
     }

     else if (src_port < dst_port)
     {
          return SESSION_DIR_S2C;
     }
     else
     {
          // if port is equal, first packet is C2S
          return SESSION_DIR_C2S;
     }
}

// OK
static enum session_dir judge_direction_by_session(const struct session *sess, const struct tuple6 *key)
{
     if (tuple6_cmp(session_get0_tuple6(sess), key) == 0)
     {
          return session_get_tuple6_dir(sess);
     }
     else
     {
          return (session_get_tuple6_dir(sess) == SESSION_DIR_C2S ? SESSION_DIR_S2C : SESSION_DIR_C2S);
     }
}

/******************************************************************************
 * timeout callback
 ******************************************************************************/

// OK
static void tcp_init_timeout_cb(struct session *sess, void *arg)
{
     SESSION_LOG_DEBUG("run tcp_init_timeout_cb on session %lu", session_get_id(sess));
     struct session_manager *mgr = (struct session_manager *)arg;
     assert(mgr != NULL);

     session_manager_update_session_on_closing(mgr, sess, CLOSING_BY_TIMEOUT);
}

// OK
static void tcp_handshake_timeout_cb(struct session *sess, void *arg)
{
     SESSION_LOG_DEBUG("run tcp_handshake_timeout_cb on session %lu", session_get_id(sess));
     struct session_manager *mgr = (struct session_manager *)arg;
     assert(mgr != NULL);

     session_manager_update_session_on_closing(mgr, sess, CLOSING_BY_TIMEOUT);
}

// OK
static void tcp_data_timeout_cb(struct session *sess, void *arg)
{
     SESSION_LOG_DEBUG("run tcp_data_timeout_cb on session %lu", session_get_id(sess));
     struct session_manager *mgr = (struct session_manager *)arg;
     assert(mgr != NULL);

     session_manager_update_session_on_closing(mgr, sess, CLOSING_BY_TIMEOUT);
}

// OK
static void tcp_half_closed_timeout_cb(struct session *sess, void *arg)
{
     SESSION_LOG_DEBUG("run tcp_half_closed_timeout_cb on session %lu", session_get_id(sess));
     struct session_manager *mgr = (struct session_manager *)arg;
     assert(mgr != NULL);

     session_manager_update_session_on_closing(mgr, sess, CLOSING_BY_TIMEOUT);
}

// OK
static void udp_data_timeout_cb(struct session *sess, void *arg)
{
     SESSION_LOG_DEBUG("run udp_data_timeout_cb on session %lu", session_get_id(sess));
     struct session_manager *mgr = (struct session_manager *)arg;
     assert(mgr != NULL);

     session_manager_update_session_on_closing(mgr, sess, CLOSING_BY_TIMEOUT);
}

/******************************************************************************
 * update session
 ******************************************************************************/

/*
on opening update session
[*] session_init
[*] session_set_id
[*] session_set_tuple6
[*] session_set_tuple6_dir
[*] session_set_type
[*] session_set_create_time
[*] session_set_state

on packet update session
[*] session_inc_c2s_metrics
[*] session_inc_s2c_metrics
[*] session_set_c2s_1st_pkt
[*] session_set_s2c_1st_pkt
[*] session_set_c2s_1st_pkt_md
[*] session_set_s2c_1st_pkt_md
[*] session_set0_cur_pkt
[*] session_set_cur_dir
[*] session_set_last_time
    session_set_state
    session_set_dup_traffic_flag

on closing update session
[*] session_set_state
[*] session_set_closing_reasion
*/

// OK
static void session_manager_update_session_state(struct session_manager *mgr, struct session *sess, enum session_state state)
{
     // session state not change
     if (session_get_state(sess) == state)
     {
          return;
     }

     enum session_type type = session_get_type(sess);
     if (type == SESSION_TYPE_TCP)
     {
          // handle old state
          switch (session_get_state(sess))
          {
          case SESSION_STATE_OPENING:
               mgr->tcp_opening_sess_num--;
               break;
          case SESSION_STATE_ACTIVE:
               mgr->tcp_active_sess_num--;
               break;
          case SESSION_STATE_CLOSING:
               mgr->tcp_closing_sess_num--;
               break;
          case SESSION_STATE_CLOSED:
               /* void */
               break;
          default:
               break;
          }

          // handle new state
          switch (state)
          {
          case SESSION_STATE_OPENING:
               mgr->tcp_opening_sess_num++;
               mgr->tcp_sess_num++;
               break;
          case SESSION_STATE_ACTIVE:
               mgr->tcp_active_sess_num++;
               break;
          case SESSION_STATE_CLOSING:
               mgr->tcp_closing_sess_num++;
               break;
          case SESSION_STATE_CLOSED:
               mgr->tcp_sess_num--;
               break;
          default:
               break;
          }
     }
     else if (type == SESSION_TYPE_UDP)
     {
          // handle old state
          switch (session_get_state(sess))
          {
          case SESSION_STATE_OPENING:
               mgr->udp_opening_sess_num--;
               break;
          case SESSION_STATE_ACTIVE:
               mgr->udp_active_sess_num--;
               break;
          case SESSION_STATE_CLOSING:
               mgr->udp_closing_sess_num--;
               break;
          case SESSION_STATE_CLOSED:
               /* void */
               break;
          default:
               break;
          }

          // handle new state
          switch (state)
          {
          case SESSION_STATE_OPENING:
               mgr->udp_opening_sess_num++;
               mgr->udp_sess_num++;
               break;
          case SESSION_STATE_ACTIVE:
               mgr->udp_active_sess_num++;
               break;
          case SESSION_STATE_CLOSING:
               mgr->udp_closing_sess_num++;
               break;
          case SESSION_STATE_CLOSED:
               mgr->udp_sess_num--;
               break;
          default:
               break;
          }
     }

     session_set_state(sess, state);
}

// OK
static inline void session_manager_update_session_on_opening(struct session_manager *mgr, struct session *sess, const struct tuple6 *key, enum session_dir curr_dir)
{
     session_init(sess);
     session_set_id(sess, session_manager_alloc_session_id());
     session_set_tuple6(sess, key);
     session_set_tuple6_dir(sess, curr_dir);
     if (key->ip_proto == IPPROTO_UDP)
     {
          session_set_type(sess, SESSION_TYPE_UDP);
     }
     else if (key->ip_proto == IPPROTO_TCP)
     {
          session_set_type(sess, SESSION_TYPE_TCP);
     }
     session_set_create_time(sess, timestamp_get_msec());
     session_manager_update_session_state(mgr, sess, SESSION_STATE_OPENING);
}

// OK
static inline void session_manager_update_session_on_packet(struct session_manager *mgr, struct session *sess, const struct packet *pkt, enum session_dir curr_dir)
{
     uint64_t len = packet_get_raw_len(pkt);
     const struct metadata *md = packet_get0_metadata(pkt);
     if (curr_dir == SESSION_DIR_C2S)
     {
          session_inc_c2s_metrics(sess, 1, len);
     }
     else if (curr_dir == SESSION_DIR_S2C)
     {
          session_inc_s2c_metrics(sess, 1, len);
     }
     session_update_packet_exdata(sess, pkt, curr_dir);
     session_update_pktmeta_exdata(sess, md, curr_dir);
     session_set0_cur_pkt(sess, pkt);
     session_set_cur_dir(sess, curr_dir);
     session_set_last_time(sess, timestamp_get_msec());
}

// OK
static inline void session_manager_update_session_on_closing(struct session_manager *mgr, struct session *sess, enum closing_reasion reasion)
{
     session_manager_update_session_state(mgr, sess, SESSION_STATE_CLOSING);
     session_set_closing_reasion(sess, reasion);
     session_timer_del_session(mgr->sess_timer, sess);
}

/******************************************************************************
 * handle session
 ******************************************************************************/

// OK
static void session_manager_free_session(struct session_manager *mgr, struct session *sess)
{
     if (sess)
     {
          session_manager_update_session_state(mgr, sess, SESSION_STATE_CLOSED);
          for (uint8_t i = 0; i < EX_DATA_MAX_COUNT; i++)
          {
               session_free_ex_data(sess, i);
          }
          session_table_del_session(mgr->tcp_sess_table, session_get0_tuple6(sess));
          session_timer_del_session(mgr->sess_timer, sess);
          session_set0_cur_pkt(sess, NULL);
          session_set_cur_dir(sess, SESSION_DIR_NONE);
          session_pool_free(mgr->sess_pool, sess);
          sess = NULL;
     }
}

// OK
static void session_manager_recycle_session(struct session_manager *mgr)
{
     while (1)
     {
          struct session *sess = session_queue_pop(mgr->sess_toclosed_queue);
          if (sess == NULL)
          {
               break;
          }
          session_manager_free_session(mgr, sess);
     }
}

// OK
static void session_manager_evicte_session(struct session_manager *mgr, struct session *sess)
{
     session_manager_update_session_on_closing(mgr, sess, CLOSING_BY_EVICTED);

     session_table_del_session(mgr->tcp_sess_table, session_get0_tuple6(sess));
     session_queue_push(mgr->sess_evicted_queue, sess);

     if (session_get_type(sess) == SESSION_TYPE_UDP)
     {
          eviction_filter_add(mgr->udp_eviction_filter, session_get0_1st_pkt(sess));
     }
}

// OK
static struct session *session_manager_handle_tcp_new_session(struct session_manager *mgr, const struct packet *pkt, const struct tuple6 *key)
{
     const struct layer_record *tcp_layer = packet_get_innermost_layer(pkt, LAYER_TYPE_TCP);
     if (tcp_layer == NULL)
     {
          mgr->npkts_miss_l4_proto++;
          return NULL;
     }

     const struct tcphdr *hdr = (const struct tcphdr *)tcp_layer->hdr_ptr;
     if (!tcp_hdr_has_flag_syn(hdr))
     {
          mgr->npkts_hit_tcp_evicted++;
          return NULL;
     }

     if (mgr->tcp_sess_num == mgr->config.max_tcp_session_num - 1)
     {
          struct session *evicted_sess = session_table_find_least_recently_unused_session(mgr->tcp_sess_table);
          assert(evicted_sess);
          session_manager_evicte_session(mgr, evicted_sess);
     }

     struct session *sess = session_pool_alloc(mgr->sess_pool);
     assert(sess);

     enum session_dir curr_dir = tcp_hdr_has_flag_ack(hdr) ? SESSION_DIR_S2C : SESSION_DIR_C2S;
     session_manager_update_session_on_opening(mgr, sess, key, curr_dir);
     session_manager_update_session_on_packet(mgr, sess, pkt, curr_dir);
     session_update_tcp_exdata(sess, tcp_layer, curr_dir);
     if (tcp_hdr_has_flag_ack(hdr))
     {
          session_set_expirecb(sess, tcp_handshake_timeout_cb, mgr, timestamp_get_sec() + mgr->config.tcp_timeout_handshake);
     }
     else
     {
          session_set_expirecb(sess, tcp_init_timeout_cb, mgr, timestamp_get_sec() + mgr->config.tcp_timeout_init);
     }
     session_timer_add_session(mgr->sess_timer, sess);
     dupkt_filter_add(mgr->tcp_dupkt_filter, pkt);

     return sess;
}

// OK
static struct session *session_manager_handle_tcp_old_session(struct session_manager *mgr, struct session *sess, const struct packet *pkt, const struct tuple6 *key)
{
     if (session_get_state(sess) == SESSION_STATE_CLOSING)
     {
          mgr->npkts_hit_tcp_closing++;
          return NULL;
     }

     const struct layer_record *tcp_layer = packet_get_innermost_layer(pkt, LAYER_TYPE_TCP);
     if (tcp_layer == NULL)
     {
          mgr->npkts_miss_l4_proto++;
          return NULL;
     }

     enum session_dir curr_dir = judge_direction_by_session(sess, key);
     if (session_manager_filter_session_dupkt(mgr, sess, pkt, curr_dir))
     {
          mgr->npkts_hit_tcp_dupkt++;
          session_set_dup_traffic_flag(sess, DUP_TRAFFIC_YES);
          return NULL;
     }

     uint64_t tcp_old_exdata = (uint64_t)session_get0_ex_data(sess, tcp_builtin_ex);
     session_manager_update_session_on_packet(mgr, sess, pkt, curr_dir);
     session_update_tcp_exdata(sess, tcp_layer, curr_dir);
     uint64_t tcp_curr_exdata = (uint64_t)session_get0_ex_data(sess, tcp_builtin_ex);

     // update tcp session state & timeout callback

     // TCP SYN retransmission
     if (tcp_old_exdata == TCP_SYN_RECVED && tcp_curr_exdata == TCP_SYN_RECVED)
     {
          session_manager_update_session_state(mgr, sess, SESSION_STATE_OPENING);
          session_set_expirecb(sess, tcp_init_timeout_cb, mgr, timestamp_get_sec() + mgr->config.tcp_timeout_init);
          session_timer_add_session(mgr->sess_timer, sess);

          return sess;
     }

     // TCP SYNACK retransmission & TCP S2C asymmetric
     if (tcp_old_exdata == TCP_SYNACK_RECVED && tcp_curr_exdata == TCP_SYNACK_RECVED)
     {
          session_manager_update_session_state(mgr, sess, SESSION_STATE_OPENING);
          session_set_expirecb(sess, tcp_handshake_timeout_cb, mgr, timestamp_get_sec() + mgr->config.tcp_timeout_handshake);
          session_timer_add_session(mgr->sess_timer, sess);

          return sess;
     }

     // TCP handshake success
     if (!(tcp_old_exdata & TCP_SYNACK_RECVED) && (tcp_curr_exdata & TCP_SYNACK_RECVED))
     {
          session_manager_update_session_state(mgr, sess, SESSION_STATE_OPENING);
          session_set_expirecb(sess, tcp_handshake_timeout_cb, mgr, timestamp_get_sec() + mgr->config.tcp_timeout_handshake);
          session_timer_add_session(mgr->sess_timer, sess);

          return sess;
     }

     // TCP established
     if ((tcp_curr_exdata & TCP_C2S_DATA_RECVED) || (tcp_curr_exdata & TCP_S2C_DATA_RECVED))
     {
          session_manager_update_session_state(mgr, sess, SESSION_STATE_ACTIVE);
          session_set_expirecb(sess, tcp_data_timeout_cb, mgr, timestamp_get_sec() + mgr->config.tcp_timeout_data);
          session_timer_add_session(mgr->sess_timer, sess);

          return sess;
     }

     // TCP closing
     if ((tcp_curr_exdata & TCP_C2S_RST_RECVED) || (tcp_curr_exdata & TCP_S2C_RST_RECVED) || ((tcp_curr_exdata & TCP_C2S_FIN_RECVED) && (tcp_curr_exdata & TCP_S2C_FIN_RECVED)))
     {
          if (tcp_curr_exdata & TCP_C2S_RST_RECVED)
          {
               session_set_closing_reasion(sess, CLOSING_BY_CLIENT_RST);
          }
          else if (tcp_curr_exdata & TCP_S2C_RST_RECVED)
          {
               session_set_closing_reasion(sess, CLOSING_BY_SERVER_RST);
          }
          session_manager_update_session_state(mgr, sess, SESSION_STATE_CLOSING);

          return sess;
     }

     // TCP half closed
     if ((tcp_curr_exdata & TCP_C2S_FIN_RECVED) || (tcp_curr_exdata & TCP_S2C_FIN_RECVED))
     {
          if (tcp_curr_exdata & TCP_C2S_FIN_RECVED)
          {
               session_set_closing_reasion(sess, CLOSING_BY_CLIENT_FIN);
          }
          else if (tcp_curr_exdata & TCP_S2C_FIN_RECVED)
          {
               session_set_closing_reasion(sess, CLOSING_BY_SERVER_FIN);
          }
          // session still is active
          session_manager_update_session_state(mgr, sess, SESSION_STATE_ACTIVE);
          session_set_expirecb(sess, tcp_half_closed_timeout_cb, mgr, timestamp_get_sec() + mgr->config.tcp_timeout_half_closed);
          session_timer_add_session(mgr->sess_timer, sess);

          return sess;
     }

     assert(0);
     session_set_expirecb(sess, tcp_data_timeout_cb, mgr, timestamp_get_sec() + mgr->config.tcp_timeout_data);
     session_timer_add_session(mgr->sess_timer, sess);

     return sess;
}

// OK
static struct session *session_manager_handle_udp_new_session(struct session_manager *mgr, const struct packet *pkt, const struct tuple6 *key)
{
     if (eviction_filter_lookup(mgr->udp_eviction_filter, pkt))
     {
          mgr->npkts_hit_udp_evicted++;
          return NULL;
     }

     if (mgr->udp_sess_num == mgr->config.max_udp_session_num - 1)
     {
          struct session *evicted_sess = session_table_find_least_recently_unused_session(mgr->udp_sess_table);
          assert(evicted_sess);
          session_manager_evicte_session(mgr, evicted_sess);
     }

     struct session *sess = session_pool_alloc(mgr->sess_pool);
     assert(sess);

     enum session_dir curr_dir = judge_direction_by_tuple6(key);
     session_manager_update_session_on_opening(mgr, sess, key, curr_dir);
     session_manager_update_session_on_packet(mgr, sess, pkt, curr_dir);
     session_manager_update_session_state(mgr, sess, SESSION_STATE_ACTIVE); // change session state to active
     session_update_udp_exdata(sess, NULL, curr_dir);
     session_set_expirecb(sess, udp_data_timeout_cb, mgr, timestamp_get_sec() + mgr->config.udp_timeout_data);
     session_timer_add_session(mgr->sess_timer, sess);

     return sess;
}

// OK
static struct session *session_manager_handle_udp_old_session(struct session_manager *mgr, struct session *sess, const struct packet *pkt, const struct tuple6 *key)
{
     if (session_get_state(sess) == SESSION_STATE_CLOSING)
     {
          mgr->npkts_hit_udp_evicted++;
          return NULL;
     }

     enum session_dir curr_dir = judge_direction_by_session(sess, key);
     session_manager_update_session_on_packet(mgr, sess, pkt, curr_dir);
     session_manager_update_session_state(mgr, sess, SESSION_STATE_ACTIVE);
     session_update_udp_exdata(sess, NULL, curr_dir);
     session_set_expirecb(sess, udp_data_timeout_cb, mgr, timestamp_get_sec() + mgr->config.udp_timeout_data);
     session_timer_add_session(mgr->sess_timer, sess);

     return sess;
}

/******************************************************************************
 * public API
 ******************************************************************************/

// OK
struct session_manager *session_manager_create(struct session_manager_config *config)
{
     if (session_manager_check_config(config))
     {
          return NULL;
     }

     struct session_manager *mgr = (struct session_manager *)calloc(1, sizeof(struct session_manager));
     if (mgr == NULL)
     {
          return NULL;
     }

     memcpy(&mgr->config, config, sizeof(struct session_manager_config));
     mgr->sess_pool = session_pool_create(mgr->config.max_tcp_session_num + mgr->config.max_udp_session_num);
     if (mgr->sess_pool == NULL)
     {
          goto error;
     }

     mgr->tcp_sess_table = session_table_create();
     if (mgr->tcp_sess_table == NULL)
     {
          goto error;
     }

     mgr->udp_sess_table = session_table_create();
     if (mgr->udp_sess_table == NULL)
     {
          goto error;
     }

     mgr->sess_timer = session_timer_create();
     if (mgr->sess_timer == NULL)
     {
          goto error;
     }

     mgr->sess_evicted_queue = session_queue_create();
     if (mgr->sess_evicted_queue == NULL)
     {
          goto error;
     }

     mgr->sess_toclosed_queue = session_queue_create();
     if (mgr->sess_toclosed_queue == NULL)
     {
          goto error;
     }

     mgr->tcp_dupkt_filter = dupkt_filter_create(mgr->config.tcp_dupkt_filter_enable, mgr->config.tcp_dupkt_filter_capacity, mgr->config.tcp_dupkt_filter_error_rate, mgr->config.tcp_dupkt_filter_timeout);
     if (mgr->tcp_dupkt_filter == NULL)
     {
          goto error;
     }

     mgr->udp_eviction_filter = eviction_filter_create(mgr->config.udp_eviction_filter_enable, mgr->config.udp_eviction_filter_capacity, mgr->config.udp_eviction_filter_error_rate, mgr->config.udp_eviction_filter_timeout);
     if (mgr->udp_eviction_filter == NULL)
     {
          goto error;
     }

     // init builtin session ex data index
     tcp_builtin_ex = session_get_ex_new_index("tcp_builtin_ex", NULL, NULL);
     udp_builtin_ex = session_get_ex_new_index("udp_builtin_ex", NULL, NULL);
     c2s_1st_md_ex = session_get_ex_new_index("c2s_1st_md_ex", session_free_pktmeta_exdata, NULL);
     s2c_1st_md_ex = session_get_ex_new_index("s2c_1st_md_ex", session_free_pktmeta_exdata, NULL);
     c2s_1st_pkt_ex = session_get_ex_new_index("c2s_1st_pkt_ex", session_free_packet_exdata, NULL);
     s2c_1st_pkt_ex = session_get_ex_new_index("s2c_1st_pkt_ex", session_free_packet_exdata, NULL);

     return mgr;

error:
     session_manager_destroy(mgr);
     return NULL;
}

// OK
void session_manager_destroy(struct session_manager *mgr)
{
     if (mgr)
     {
          eviction_filter_destroy(mgr->udp_eviction_filter);
          dupkt_filter_destroy(mgr->tcp_dupkt_filter);
          session_queue_destroy(mgr->sess_toclosed_queue);
          session_queue_destroy(mgr->sess_evicted_queue);
          session_timer_destroy(mgr->sess_timer);
          session_table_destroy(mgr->udp_sess_table);
          session_table_destroy(mgr->tcp_sess_table);
          session_pool_destroy(mgr->sess_pool);
          free(mgr);
          mgr = NULL;
     }
}

// OK
// Only use the packet six-tuple to find the session, not update it
struct session *session_manager_lookup_sesssion(struct session_manager *mgr, const struct packet *pkt)
{
     struct tuple6 key;
     memset(&key, 0, sizeof(struct tuple6));
     if (packet_get_innermost_tuple6(pkt, &key))
     {
          return NULL;
     }

     if (key.ip_proto == IPPROTO_UDP)
     {
          return session_table_find_session(mgr->udp_sess_table, &key);
     }
     else if (key.ip_proto == IPPROTO_TCP)
     {
          return session_table_find_session(mgr->tcp_sess_table, &key);
     }
     else
     {
          return NULL;
     }
}

// OK
/*
 * Return NULL in the following cases:
 *      1.not a TCP or UDP packet
 *      2.TCP packet miss session but no syn packet seen
 *      3.TCP duplicate packet
 *      4.TCP discards packets
 *      5.UDP evict packet
 * pakcet will not update the session and needs to be fast forwarded
 */
struct session *session_manager_update_session(struct session_manager *mgr, const struct packet *pkt)
{
     assert(session_manager_get_evicted_session(mgr) == NULL);

     session_manager_recycle_session(mgr);

     struct tuple6 key;
     memset(&key, 0, sizeof(struct tuple6));
     if (packet_get_innermost_tuple6(pkt, &key))
     {
          mgr->npkts_miss_l4_proto++;
          return NULL;
     }

     struct session *sess = NULL;
     if (key.ip_proto == IPPROTO_UDP)
     {
          sess = session_table_find_session(mgr->udp_sess_table, &key);
          if (sess)
          {
               return session_manager_handle_udp_old_session(mgr, sess, pkt, &key);
          }
          else
          {
               return session_manager_handle_udp_new_session(mgr, pkt, &key);
          }
     }
     else if (key.ip_proto == IPPROTO_TCP)
     {
          sess = session_table_find_session(mgr->tcp_sess_table, &key);
          if (sess)
          {
               return session_manager_handle_tcp_old_session(mgr, sess, pkt, &key);
          }
          else
          {
               return session_manager_handle_tcp_new_session(mgr, pkt, &key);
          }
     }
     else
     {
          return NULL;
     }
}

// OK
struct session *session_manager_get_expired_session(struct session_manager *mgr)
{
     session_manager_recycle_session(mgr);

     SESSION_LOG_DEBUG("current timestamp: %lu s", timestamp_get_sec());
     struct session *sess = session_timer_expire_session(mgr->sess_timer, timestamp_get_msec());
     if (sess == NULL)
     {
          session_run_expirecb(sess);
          session_queue_push(mgr->sess_toclosed_queue, sess);
     }

     return sess;
}

// OK
struct session *session_manager_get_evicted_session(struct session_manager *mgr)
{
     session_manager_recycle_session(mgr);

     struct session *sess = session_queue_pop(mgr->sess_evicted_queue);
     if (sess)
     {
          session_queue_push(mgr->sess_toclosed_queue, sess);
     }

     return sess;
}

// OK
// return interval (seconds) to next required update, return 0 if no session
uint64_t session_manager_get_expire_interval(struct session_manager *mgr)
{
     return session_timer_next_expire_interval(mgr->sess_timer);
}

// OK
uint64_t session_manager_get_session_number(struct session_manager *mgr, enum session_type type, enum session_state state)
{
     if (type == SESSION_TYPE_TCP)
     {
          switch (state)
          {
          case SESSION_STATE_OPENING:
               return mgr->tcp_opening_sess_num;
          case SESSION_STATE_ACTIVE:
               return mgr->tcp_active_sess_num;
          case SESSION_STATE_CLOSING:
               return mgr->tcp_closing_sess_num;
          default:
               return 0;
          }
     }
     else if (type == SESSION_TYPE_UDP)
     {
          switch (state)
          {
          case SESSION_STATE_OPENING:
               return mgr->udp_opening_sess_num;
          case SESSION_STATE_ACTIVE:
               return mgr->udp_active_sess_num;
          case SESSION_STATE_CLOSING:
               return mgr->udp_closing_sess_num;
          default:
               return 0;
          }
     }

     return 0;
}