stellar-stellar/src/session/session_manager.cpp

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

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

struct session_manager
{
     struct session_pool *sess_pool;
     struct session_table *sess_table;
     struct session_timer *sess_timer;
     struct session_queue *evicted_sess;

     session_event_cb event_cb;
     void *arg;

     // timeout config
     uint64_t timeout_to_closing;
     uint64_t timeout_to_closed;

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

     uint64_t udp_opening_sess_num;
     uint64_t udp_closing_sess_num;
     uint64_t udp_active_sess_num;
};

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

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

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

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

/******************************************************************************
 * session manager counter
 ******************************************************************************/

static void update_counter_on_opening(struct session_manager *mgr, struct session *sess)
{
     if (session_get_state(sess) == SESSION_STATE_INIT)
     {
          if (session_get_type(sess) == SESSION_TYPE_TCP)
          {
               mgr->tcp_opening_sess_num++;
          }
          else
          {
               mgr->udp_opening_sess_num++;
          }
     }
}

static void update_counter_on_active(struct session_manager *mgr, struct session *sess)
{
     if (session_get_state(sess) == SESSION_STATE_OPENING)
     {
          if (session_get_type(sess) == SESSION_TYPE_TCP)
          {
               mgr->tcp_opening_sess_num--;
               mgr->tcp_active_sess_num++;
          }
          else
          {
               mgr->udp_opening_sess_num--;
               mgr->udp_active_sess_num++;
          }
     }
}

static void update_counter_on_closing(struct session_manager *mgr, struct session *sess)
{
     if (session_get_state(sess) == SESSION_STATE_OPENING)
     {
          if (session_get_type(sess) == SESSION_TYPE_TCP)
          {
               mgr->tcp_opening_sess_num--;
               mgr->tcp_closing_sess_num++;
          }
          else
          {
               mgr->udp_opening_sess_num--;
               mgr->udp_closing_sess_num++;
          }

          return;
     }

     if (session_get_state(sess) == SESSION_STATE_ACTIVE)
     {
          if (session_get_type(sess) == SESSION_TYPE_TCP)
          {
               mgr->tcp_active_sess_num--;
               mgr->tcp_closing_sess_num++;
          }
          else
          {
               mgr->udp_active_sess_num--;
               mgr->udp_closing_sess_num++;
          }

          return;
     }
}

static void update_counter_on_closed(struct session_manager *mgr, struct session *sess)
{
     if (session_get_state(sess) == SESSION_STATE_CLOSING)
     {
          if (session_get_type(sess) == SESSION_TYPE_TCP)
          {
               mgr->tcp_closing_sess_num--;
          }
          else
          {
               mgr->udp_closing_sess_num--;
          }
     }
}

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

static enum session_dir judge_direction_by_tuple6(const struct tuple6 *key)
{
     // big port is client
     if (ntohs(key->src_port) > ntohs(key->dst_port))
     {
          return SESSION_DIR_C2S;
     }
     else
     {
          return SESSION_DIR_S2C;
     }
}

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
     {
          if (session_get_tuple6_dir(sess) == SESSION_DIR_C2S)
          {
               return SESSION_DIR_S2C;
          }
          else
          {
               return SESSION_DIR_C2S;
          }
     }
}

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

void session_manager_update_session_timer(struct session_manager *mgr, struct session *sess, session_expire_cb cb, uint64_t timeout_ms)
{
     session_timer_del_session(mgr->sess_timer, sess);
     session_set_expirecb(sess, cb, mgr, timestamp_get_msec() + timeout_ms);
     session_timer_add_session(mgr->sess_timer, sess);
}

/******************************************************************************
 * expire callback
 ******************************************************************************/

static void session_to_closed(struct session *sess, void *arg)
{
     SESSION_MANAGER_LOG_DEBUG("session %lu closing expire, free session", session_get_id(sess));
     struct session_manager *mgr = (struct session_manager *)arg;
     assert(mgr != NULL);

     uint32_t event;
     while (session_pop_event(sess, &event))
     {
     }

     update_counter_on_closed(mgr, sess);
     session_set_state(sess, SESSION_STATE_CLOSED);
     session_set0_cur_pkt(sess, NULL);
     session_set_cur_dir(sess, SESSION_DIR_NONE);
     for (uint8_t i = 0; i < EX_DATA_MAX_COUNT; i++)
     {
          session_free_ex_data(sess, i);
     }
     session_table_delete_session(mgr->sess_table, session_get0_tuple6(sess));
     session_timer_del_session(mgr->sess_timer, sess);
     session_pool_free(mgr->sess_pool, sess);
}

static void session_to_closing(struct session *sess, void *arg)
{
     SESSION_MANAGER_LOG_DEBUG("session %lu packet expire, trigger closing event", session_get_id(sess));
     struct session_manager *mgr = (struct session_manager *)arg;
     assert(mgr != NULL);

     update_counter_on_closing(mgr, sess);
     session_set_state(sess, SESSION_STATE_CLOSING);
     session_push_event(sess, SESSION_EVENT_CLOSING);
     session_manager_update_session_timer(mgr, sess, session_to_closed, mgr->timeout_to_closed);
}

/******************************************************************************
 * session ex data
 ******************************************************************************/

static int tcp_need_closing(uint64_t state)
{
     if ((state & TCP_C2S_FIN_RECVED) && (state & TCP_S2C_FIN_RECVED))
     {
          return 1;
     }

     if (state & TCP_C2S_RST_RECVED)
     {
          return 1;
     }

     if (state & TCP_S2C_RST_RECVED)
     {
          return 1;
     }

     return 0;
}

static int tcp_need_active(uint64_t state)
{
     if ((state & TCP_C2S_PAYLOAD_RECVED) || (state & TCP_S2C_PAYLOAD_RECVED))
     {
          return 1;
     }

     return 0;
}

static void update_session_base(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);
          if (session_get0_ex_data(sess, c2s_1st_md_ex) == NULL)
          {
               session_set_ex_data(sess, c2s_1st_md_ex, metadata_dup(md));
          }
          if (session_get0_ex_data(sess, c2s_1st_pkt_ex) == NULL)
          {
               session_set_ex_data(sess, c2s_1st_pkt_ex, packet_dup(pkt));
          }
     }
     else
     {
          session_inc_s2c_metrics(sess, 1, len);
          if (session_get0_ex_data(sess, s2c_1st_md_ex) == NULL)
          {
               session_set_ex_data(sess, s2c_1st_md_ex, metadata_dup(md));
          }
          if (session_get0_ex_data(sess, s2c_1st_pkt_ex) == NULL)
          {
               session_set_ex_data(sess, s2c_1st_pkt_ex, packet_dup(pkt));
          }
     }
     session_set_last_time(sess, timestamp_get_msec());
     session_set0_cur_pkt(sess, pkt);
     session_set_cur_dir(sess, curr_dir);
}

static void update_tcp_ex_data(struct session *sess, const struct packet *pkt, enum session_dir curr_dir)
{
     const struct layer_record *tcp_layer = packet_get_innermost_layer(pkt, LAYER_TYPE_TCP);
     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_rst(hdr))
     {
          if (curr_dir == SESSION_DIR_C2S)
          {
               state |= TCP_C2S_RST_RECVED;
               session_set_ex_data(sess, tcp_builtin_ex, (void *)(state));
          }
          else
          {
               state |= TCP_S2C_RST_RECVED;
               session_set_ex_data(sess, tcp_builtin_ex, (void *)(state));
          }
     }

     if (tcp_hdr_has_flag_fin(hdr))
     {
          if (curr_dir == SESSION_DIR_C2S)
          {
               state |= TCP_C2S_FIN_RECVED;
               session_set_ex_data(sess, tcp_builtin_ex, (void *)(state));
          }
          else
          {
               state |= TCP_S2C_FIN_RECVED;
               session_set_ex_data(sess, tcp_builtin_ex, (void *)(state));
          }
     }

     if (tcp_hdr_has_flag_syn(hdr))
     {
          if (tcp_hdr_has_flag_ack(hdr))
          {
               state |= TCP_SYNACK_RECVED;
               session_set_ex_data(sess, tcp_builtin_ex, (void *)(state));
          }
          else
          {
               state |= TCP_SYN_RECVED;
               session_set_ex_data(sess, tcp_builtin_ex, (void *)(state));
          }
     }

     if (tcp_layer->pld_len > 0)
     {
          if (curr_dir == SESSION_DIR_C2S)
          {
               state |= TCP_C2S_PAYLOAD_RECVED;
               session_set_ex_data(sess, tcp_builtin_ex, (void *)(state));
          }
          else
          {
               state |= TCP_S2C_PAYLOAD_RECVED;
               session_set_ex_data(sess, tcp_builtin_ex, (void *)(state));
          }
     }
}

static void update_udp_ex_data(struct session *sess, const struct packet *pkt, enum session_dir curr_dir)
{
     uint64_t state = (uint64_t)session_get0_ex_data(sess, udp_builtin_ex);
     if (curr_dir == SESSION_DIR_C2S)
     {
          session_set_ex_data(sess, udp_builtin_ex, (void *)(state | UDP_C2S_RECVED));
     }
     else
     {
          session_set_ex_data(sess, udp_builtin_ex, (void *)(state | UDP_S2C_RECVED));
     }
}

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

// return  0: success
// return -1: tcp not syn packet, discard
static int handle_tcp_new_session(struct session_manager *mgr, struct tuple6 *key, struct session *sess, const struct packet *pkt)
{
     const struct tcphdr *hdr = (const struct tcphdr *)packet_get_tcp_hdr_ptr(pkt);
     if (!tcp_hdr_has_flag_syn(hdr))
     {
          // not syn packet, discard
          return -1;
     }

     enum session_dir curr_dir = SESSION_DIR_NONE;

     session_init(sess);
     // syn packet
     if (!tcp_hdr_has_flag_ack(hdr))
     {
          curr_dir = SESSION_DIR_C2S;
          session_set_ex_data(sess, tcp_builtin_ex, (void *)TCP_SYN_RECVED);
     }
     // syn ack packet
     else
     {
          curr_dir = SESSION_DIR_S2C;
          session_set_ex_data(sess, tcp_builtin_ex, (void *)TCP_SYNACK_RECVED);
     }

     session_set_id(sess, alloc_session_id());
     session_set_tuple6(sess, key);
     session_set_tuple6_dir(sess, curr_dir);

     session_set_type(sess, SESSION_TYPE_TCP);
     update_counter_on_opening(mgr, sess);
     session_set_state(sess, SESSION_STATE_OPENING);

     session_set_create_time(sess, timestamp_get_msec());
     update_session_base(sess, pkt, curr_dir);

     session_push_event(sess, SESSION_EVENT_OPENING);
     session_push_event(sess, SESSION_EVENT_PACKET);
     session_manager_update_session_timer(mgr, sess, session_to_closing, mgr->timeout_to_closing);

     return 0;
}

// always return 0
static int handle_udp_new_session(struct session_manager *mgr, struct tuple6 *key, struct session *sess, const struct packet *pkt)
{
     enum session_dir curr_dir = judge_direction_by_tuple6(key);

     session_init(sess);
     update_udp_ex_data(sess, pkt, curr_dir);
     session_set_id(sess, alloc_session_id());
     session_set_tuple6(sess, key);
     session_set_tuple6_dir(sess, curr_dir);

     /*
      * when a UDP Session is created, the Opening and active events are triggered,
      * (the plugin is called twice by the opening/active events in turn),
      * and the state of the UDP session is directly switched to the active state.
      */
     session_set_type(sess, SESSION_TYPE_UDP);
     update_counter_on_opening(mgr, sess);
     session_set_state(sess, SESSION_STATE_OPENING);
     update_counter_on_active(mgr, sess);
     session_set_state(sess, SESSION_STATE_ACTIVE);

     session_set_create_time(sess, timestamp_get_msec());
     update_session_base(sess, pkt, curr_dir);

     session_push_event(sess, SESSION_EVENT_OPENING);
     session_push_event(sess, SESSION_EVENT_PACKET);
     session_manager_update_session_timer(mgr, sess, session_to_closing, mgr->timeout_to_closing);

     return 0;
}

static void handle_tcp_old_session(struct session_manager *mgr, struct tuple6 *key, struct session *sess, const struct packet *pkt)
{
     enum session_dir curr_dir = judge_direction_by_session(sess, key);
     update_tcp_ex_data(sess, pkt, curr_dir);
     update_session_base(sess, pkt, curr_dir);

     uint64_t state = (uint64_t)session_get0_ex_data(sess, tcp_builtin_ex);
     if (tcp_need_closing(state))
     {
          update_counter_on_closing(mgr, sess);
          session_set_state(sess, SESSION_STATE_CLOSING);
          session_push_event(sess, SESSION_EVENT_PACKET);
          session_push_event(sess, SESSION_EVENT_CLOSING);
          session_manager_update_session_timer(mgr, sess, session_to_closed, mgr->timeout_to_closed);
          return;
     }

     if (tcp_need_active(state))
     {
          update_counter_on_active(mgr, sess);
          session_set_state(sess, SESSION_STATE_ACTIVE);
          session_push_event(sess, SESSION_EVENT_PACKET);
          session_manager_update_session_timer(mgr, sess, session_to_closing, mgr->timeout_to_closing);
          return;
     }
}

static void handle_udp_old_session(struct session_manager *mgr, struct tuple6 *key, struct session *sess, const struct packet *pkt)
{
     enum session_dir curr_dir = judge_direction_by_session(sess, key);
     update_udp_ex_data(sess, pkt, curr_dir);
     update_session_base(sess, pkt, curr_dir);

     update_counter_on_active(mgr, sess);
     session_set_state(sess, SESSION_STATE_ACTIVE);
     session_push_event(sess, SESSION_EVENT_PACKET);
     session_manager_update_session_timer(mgr, sess, session_to_closing, mgr->timeout_to_closing);
}

// return  0: success
// return -1: tcp not syn packet, discard
static int handle_new_session(struct session_manager *mgr, struct tuple6 *key, struct session *sess, const struct packet *pkt)
{
     if (key->ip_proto == IPPROTO_TCP)
     {
          return handle_tcp_new_session(mgr, key, sess, pkt);
     }
     else
     {
          return handle_udp_new_session(mgr, key, sess, pkt);
     }
}

static void handle_old_session(struct session_manager *mgr, struct tuple6 *key, struct session *sess, const struct packet *pkt)
{
     if (session_get_state(sess) == SESSION_STATE_CLOSING)
     {
          return;
     }

     if (key->ip_proto == IPPROTO_TCP)
     {
          handle_tcp_old_session(mgr, key, sess, pkt);
     }
     else
     {
          handle_udp_old_session(mgr, key, sess, pkt);
     }
}

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

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

     mgr->sess_pool = session_pool_create(max_session_num);
     if (mgr->sess_pool == NULL)
     {
          goto error;
     }

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

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

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

     mgr->timeout_to_closed = 2 * 1000;
     mgr->timeout_to_closing = 5 * 1000;

     mgr->tcp_opening_sess_num = 0;
     mgr->tcp_closing_sess_num = 0;
     mgr->tcp_active_sess_num = 0;

     mgr->udp_opening_sess_num = 0;
     mgr->udp_closing_sess_num = 0;
     mgr->udp_active_sess_num = 0;

     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", metadata_ex_free_cb, NULL);
     s2c_1st_md_ex = session_get_ex_new_index("s2c_1st_md_ex", metadata_ex_free_cb, NULL);
     c2s_1st_pkt_ex = session_get_ex_new_index("c2s_1st_pkt_ex", packet_ex_free_cb, NULL);
     s2c_1st_pkt_ex = session_get_ex_new_index("s2c_1st_pkt_ex", packet_ex_free_cb, NULL);

     return mgr;

error:
     session_manager_destroy(mgr);
     return NULL;
}

void session_manager_destroy(struct session_manager *mgr)
{
     if (mgr)
     {
          session_queue_destroy(mgr->evicted_sess);
          session_timer_destroy(mgr->sess_timer);
          session_table_destroy(mgr->sess_table);
          session_pool_destroy(mgr->sess_pool);
          free(mgr);
          mgr = NULL;
     }
}

void session_manager_set_session_eventcb(struct session_manager *mgr, session_event_cb cb, void *arg)
{
     mgr->event_cb = cb;
     mgr->arg = arg;
}

void session_manager_set_timeout_toclosing(struct session_manager *mgr, uint64_t timeout_ms)
{
     mgr->timeout_to_closing = timeout_ms;
}

void session_manager_set_timeout_toclosed(struct session_manager *mgr, uint64_t timeout_ms)
{
     mgr->timeout_to_closed = timeout_ms;
}

struct session *session_manager_lookup(struct session_manager *mgr, const struct packet *pkt)
{
     struct tuple6 key;
     if (packet_get_innermost_tuple6(pkt, &key) == -1)
     {
          return NULL;
     }

     return session_table_find_session(mgr->sess_table, &key);
}

// return null: Invalid tuple6 or tcp first packet is not syn
struct session *session_manager_update(struct session_manager *mgr, const struct packet *pkt)
{
     struct tuple6 key;
     if (packet_get_innermost_tuple6(pkt, &key) == -1)
     {
          return NULL;
     }

     struct session *sess = session_table_find_session(mgr->sess_table, &key);
     if (sess == NULL)
     {
          if (session_pool_get_count(mgr->sess_pool) == 1)
          {
               struct session *unused_sess = session_table_find_least_recently_unused_session(mgr->sess_table);
               assert(unused_sess);

               update_counter_on_closing(mgr, unused_sess);
               session_set_state(unused_sess, SESSION_STATE_CLOSING);
               session_push_event(unused_sess, SESSION_EVENT_CLOSING);
               session_queue_push(mgr->evicted_sess, unused_sess);
               session_manager_update_session_timer(mgr, unused_sess, session_to_closed, mgr->timeout_to_closed);
          }

          sess = session_pool_alloc(mgr->sess_pool);
          assert(sess != NULL);

          // return  0: success
          // return -1: tcp not syn packet, discard
          if (handle_new_session(mgr, &key, sess, pkt) == 0)
          {
               session_table_add_session(mgr->sess_table, &key, sess);
          }
          else
          {
               session_pool_free(mgr->sess_pool, sess);
               return NULL;
          }
     }
     else
     {
          handle_old_session(mgr, &key, sess, pkt);
     }

     return sess;
}

struct session *session_manager_expire(struct session_manager *mgr)
{
     SESSION_MANAGER_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)
     {
          return NULL;
     }

     session_run_expirecb(sess);

     if (session_get_state(sess) == SESSION_STATE_CLOSED)
     {
          return NULL;
     }

     return sess;
}

static void session_dispatch(struct session_manager *mgr, struct session *sess)
{
     if (sess == NULL)
     {
          return;
     }

     uint32_t event;
     while (session_pop_event(sess, &event))
     {
          SESSION_MANAGER_LOG_DEBUG("handle \"%s\" event on session %lu", session_event_tostring((enum session_event)event), session_get_id(sess));
          if (mgr->event_cb)
          {
               mgr->event_cb(sess, event, mgr->arg);
          }
     }
     session_set0_cur_pkt(sess, NULL);
     session_set_cur_dir(sess, SESSION_DIR_NONE);
}

void session_manager_dispatch(struct session_manager *mgr, struct session *sess)
{
     // Handle sessions that are ready to be processed
     session_dispatch(mgr, sess);

     // Handle sessions that are evicted because the flow table is full
     struct session *evicted_sess = session_queue_pop(mgr->evicted_sess);
     session_dispatch(mgr, evicted_sess);
}

uint64_t session_manager_get_tcp_opening_sess_num(struct session_manager *mgr)
{
     return mgr->tcp_opening_sess_num;
}

uint64_t session_manager_get_tcp_closing_sess_num(struct session_manager *mgr)
{
     return mgr->tcp_closing_sess_num;
}

uint64_t session_manager_get_tcp_active_sess_num(struct session_manager *mgr)
{
     return mgr->tcp_active_sess_num;
}

uint64_t session_manager_get_udp_opening_sess_num(struct session_manager *mgr)
{
     return mgr->udp_opening_sess_num;
}

uint64_t session_manager_get_udp_closing_sess_num(struct session_manager *mgr)
{
     return mgr->udp_closing_sess_num;
}

uint64_t session_manager_get_udp_active_sess_num(struct session_manager *mgr)
{
     return mgr->udp_active_sess_num;
}