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

Split session_manager_dispatch into session_manager_dispatch and session_manager_expire

Browse Source
This commit is contained in:
luwenpeng
2023-12-22 18:18:21 +08:00
parent 6ed321caea
commit b353c9c824
3 changed files with 202 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files

163
src/session/session_manager.cpp
View File

@@ -17,14 +17,14 @@ struct session_manager
struct session_pool *sess_pool;
struct session_table *sess_table;
struct session_timer *sess_timer;
struct session_queue *sess_queue; // ready session queue
struct session_queue *evicted_sess;
session_event_cb event_cb;
void *arg;
// timeout config
uint64_t packet_timeout_ms;
uint64_t closing_timeout_ms;
uint64_t timeout_to_closing;
uint64_t timeout_to_closed;
// session number
uint64_t tcp_opening_sess_num;
@@ -185,15 +185,9 @@ static enum session_dir judge_direction_by_session(const struct session *sess, c
}
/******************************************************************************
* update session event and timer
* update session timer
******************************************************************************/
void session_manager_trigger_session_event(struct session_manager *mgr, struct session *sess, uint32_t event)
{
session_push_event(sess, event);
session_queue_push(mgr->sess_queue, sess);
}
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);
@@ -205,7 +199,7 @@ void session_manager_update_session_timer(struct session_manager *mgr, struct se
* expire callback
******************************************************************************/
static void closing_expire_callback(struct session *sess, void *arg)
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;
@@ -229,7 +223,7 @@ static void closing_expire_callback(struct session *sess, void *arg)
session_pool_free(mgr->sess_pool, sess);
}
static void packet_expire_callback(struct session *sess, void *arg)
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;
@@ -237,8 +231,8 @@ static void packet_expire_callback(struct session *sess, void *arg)
update_counter_on_closing(mgr, sess);
session_set_state(sess, SESSION_STATE_CLOSING);
session_manager_trigger_session_event(mgr, sess, SESSION_EVENT_CLOSING);
session_manager_update_session_timer(mgr, sess, closing_expire_callback, mgr->closing_timeout_ms);
session_push_event(sess, SESSION_EVENT_CLOSING);
session_manager_update_session_timer(mgr, sess, session_to_closed, mgr->timeout_to_closed);
}
/******************************************************************************
@@ -426,8 +420,9 @@ static int handle_tcp_new_session(struct session_manager *mgr, struct tuple6 *ke
session_set_create_time(sess, timestamp_get_msec());
update_session_base(sess, pkt, curr_dir);
session_manager_trigger_session_event(mgr, sess, SESSION_EVENT_OPENING);
session_manager_update_session_timer(mgr, sess, packet_expire_callback, mgr->packet_timeout_ms);
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;
}
@@ -457,9 +452,9 @@ static int handle_udp_new_session(struct session_manager *mgr, struct tuple6 *ke
session_set_create_time(sess, timestamp_get_msec());
update_session_base(sess, pkt, curr_dir);
session_manager_trigger_session_event(mgr, sess, SESSION_EVENT_OPENING);
session_manager_trigger_session_event(mgr, sess, SESSION_EVENT_PACKET);
session_manager_update_session_timer(mgr, sess, packet_expire_callback, mgr->packet_timeout_ms);
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;
}
@@ -475,8 +470,9 @@ static void handle_tcp_old_session(struct session_manager *mgr, struct tuple6 *k
{
update_counter_on_closing(mgr, sess);
session_set_state(sess, SESSION_STATE_CLOSING);
session_manager_trigger_session_event(mgr, sess, SESSION_EVENT_CLOSING);
session_manager_update_session_timer(mgr, sess, closing_expire_callback, mgr->closing_timeout_ms);
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;
}
@@ -484,8 +480,8 @@ static void handle_tcp_old_session(struct session_manager *mgr, struct tuple6 *k
{
update_counter_on_active(mgr, sess);
session_set_state(sess, SESSION_STATE_ACTIVE);
session_manager_trigger_session_event(mgr, sess, SESSION_EVENT_PACKET);
session_manager_update_session_timer(mgr, sess, packet_expire_callback, mgr->packet_timeout_ms);
session_push_event(sess, SESSION_EVENT_PACKET);
session_manager_update_session_timer(mgr, sess, session_to_closing, mgr->timeout_to_closing);
return;
}
}
@@ -498,8 +494,8 @@ static void handle_udp_old_session(struct session_manager *mgr, struct tuple6 *k
update_counter_on_active(mgr, sess);
session_set_state(sess, SESSION_STATE_ACTIVE);
session_manager_trigger_session_event(mgr, sess, SESSION_EVENT_PACKET);
session_manager_update_session_timer(mgr, sess, packet_expire_callback, mgr->packet_timeout_ms);
session_push_event(sess, SESSION_EVENT_PACKET);
session_manager_update_session_timer(mgr, sess, session_to_closing, mgr->timeout_to_closing);
}
// return 0: success
@@ -563,14 +559,14 @@ struct session_manager *session_manager_create(uint64_t max_session_num)
goto error;
}
mgr->sess_queue = session_queue_create();
if (mgr->sess_queue == NULL)
mgr->evicted_sess = session_queue_create();
if (mgr->evicted_sess == NULL)
{
goto error;
}
mgr->closing_timeout_ms = 2 * 1000;
mgr->packet_timeout_ms = 5 * 1000;
mgr->timeout_to_closed = 2 * 1000;
mgr->timeout_to_closing = 5 * 1000;
mgr->tcp_opening_sess_num = 0;
mgr->tcp_closing_sess_num = 0;
@@ -598,7 +594,7 @@ void session_manager_destroy(struct session_manager *mgr)
{
if (mgr)
{
session_queue_destroy(mgr->sess_queue);
session_queue_destroy(mgr->evicted_sess);
session_timer_destroy(mgr->sess_timer);
session_table_destroy(mgr->sess_table);
session_pool_destroy(mgr->sess_pool);
@@ -613,20 +609,29 @@ void session_manager_set_session_eventcb(struct session_manager *mgr, session_ev
mgr->arg = arg;
}
void session_manager_set_packet_timeout(struct session_manager *mgr, uint64_t timeout_ms)
void session_manager_set_timeout_toclosing(struct session_manager *mgr, uint64_t timeout_ms)
{
mgr->packet_timeout_ms = timeout_ms;
mgr->timeout_to_closing = timeout_ms;
}
void session_manager_set_closing_timeout(struct session_manager *mgr, uint64_t timeout_ms)
void session_manager_set_timeout_toclosed(struct session_manager *mgr, uint64_t timeout_ms)
{
mgr->closing_timeout_ms = timeout_ms;
mgr->timeout_to_closed = timeout_ms;
}
// return NULL: discard
// * tuple6 not find
// * tcp first packet not syn
struct session *session_manager_find_session(struct session_manager *mgr, const struct packet *pkt)
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)
@@ -644,8 +649,9 @@ struct session *session_manager_find_session(struct session_manager *mgr, const
update_counter_on_closing(mgr, unused_sess);
session_set_state(unused_sess, SESSION_STATE_CLOSING);
session_manager_trigger_session_event(mgr, unused_sess, SESSION_EVENT_CLOSING);
session_manager_update_session_timer(mgr, unused_sess, closing_expire_callback, mgr->closing_timeout_ms);
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);
@@ -671,54 +677,53 @@ struct session *session_manager_find_session(struct session_manager *mgr, const
return sess;
}
void session_manager_dispatch(struct session_manager *mgr)
struct session *session_manager_expire(struct session_manager *mgr)
{
uint32_t event;
struct session *sess;
void *cb_arg = mgr->arg;
session_event_cb event_cb = mgr->event_cb;
SESSION_MANAGER_LOG_DEBUG("current timestamp: %lu s", timestamp_get_sec());
// limit expire session number
for (int i = 0; i < 100; i++)
struct session *sess = session_timer_expire_session(mgr->sess_timer, timestamp_get_msec());
if (sess == NULL)
{
sess = session_timer_expire_session(mgr->sess_timer, timestamp_get_msec());
if (sess == NULL)
{
break;
}
session_run_expirecb(sess);
return NULL;
}
while (1)
session_run_expirecb(sess);
if (session_get_state(sess) == SESSION_STATE_CLOSED)
{
// get session from ready queue
sess = session_queue_pop(mgr->sess_queue);
if (sess == NULL)
{
break;
}
return NULL;
}
while (1)
{
// get event from session
if (session_pop_event(sess, &event) == false)
{
break;
}
return sess;
}
SESSION_MANAGER_LOG_DEBUG("handle \"%s\" event on session %lu", session_event_tostring((enum session_event)event), session_get_id(sess));
if (event_cb)
{
event_cb(sess, event, cb_arg);
}
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);
};
}
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)