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

implementation session manager

Browse Source
This commit is contained in:
luwenpeng
2023-12-19 10:47:26 +08:00
parent 2e56bd810c
commit 2c26879182
15 changed files with 2269 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
src/packet/packet.cpp
View File

@@ -14,29 +14,28 @@
#define likely(expr) __builtin_expect((expr), 1) #define likely(expr) __builtin_expect((expr), 1)
#define unlikely(expr) __builtin_expect((expr), 0) #define unlikely(expr) __builtin_expect((expr), 0)
#define LOG_PACKET "PACKET" #define PACKET_LOG_DATA_INSUFFICIENCY(type) \
#define PACKET_LOG_DATA_INSUFFICIENCY(type) \ { \
{ \ PACKET_LOG_ERROR("layer: %s, data insufficiency", \
PACKET_LOG_ERROR("%s: layer: %s, data insufficiency", \ layer_type_tostring((type))); \
LOG_PACKET, layer_type_tostring((type))); \
} }
#define PACKET_LOG_UNSUPPORT_PROTO(tag, next_proto) \ #define PACKET_LOG_UNSUPPORT_PROTO(tag, next_proto) \
{ \ { \
PACKET_LOG_ERROR("%s: %s: unsupport next proto %d", \ PACKET_LOG_ERROR("%s: unsupport next proto %d", \
LOG_PACKET, (tag), (next_proto)); \ (tag), (next_proto)); \
} }
#define PACKET_LOG_UNSUPPORT_ETHPROTO(tag, next_proto) \ #define PACKET_LOG_UNSUPPORT_ETHPROTO(tag, next_proto) \
{ \ { \
PACKET_LOG_ERROR("%s: %s: unsupport next proto %d: %s", \ PACKET_LOG_ERROR("%s: unsupport next proto %d: %s", \
LOG_PACKET, (tag), (next_proto), ethproto_tostring(next_proto)); \ (tag), (next_proto), ethproto_tostring(next_proto)); \
} }
#define PACKET_LOG_UNSUPPORT_IPPROTO(tag, next_proto) \ #define PACKET_LOG_UNSUPPORT_IPPROTO(tag, next_proto) \
{ \ { \
PACKET_LOG_ERROR("%s: %s: unsupport next proto %d: %s", \ PACKET_LOG_ERROR("%s: unsupport next proto %d: %s", \
LOG_PACKET, (tag), (next_proto), ipproto_tostring(next_proto)); \ (tag), (next_proto), ipproto_tostring(next_proto)); \
} }
/****************************************************************************** /******************************************************************************
@@ -428,8 +427,8 @@ static inline struct layer_record *get_free_layer(struct packet *handler)
(_layer)->pld_ptr = (_data) + (_hdr_len); \ (_layer)->pld_ptr = (_data) + (_hdr_len); \
(_layer)->pld_len = (_len) - (_hdr_len); \ (_layer)->pld_len = (_len) - (_hdr_len); \
(_handler)->layers_used++; \ (_handler)->layers_used++; \
PACKET_LOG_DEBUG("%s: layer[%d/%d]: %s, hdr_offset: %d, hdr_ptr: %p, hdr_len: %d, pld_ptr: %p, pld_len: %d", \ PACKET_LOG_DEBUG("layer[%d/%d]: %s, hdr_offset: %d, hdr_ptr: %p, hdr_len: %d, pld_ptr: %p, pld_len: %d", \
LOG_PACKET, (_handler)->layers_used - 1, (_handler)->layers_size, layer_type_tostring((_type)), \ (_handler)->layers_used - 1, (_handler)->layers_size, layer_type_tostring((_type)), \
(_layer)->hdr_offset, (_layer)->hdr_ptr, (_layer)->hdr_len, (_layer)->pld_ptr, (_layer)->pld_len); \ (_layer)->hdr_offset, (_layer)->hdr_ptr, (_layer)->hdr_len, (_layer)->pld_ptr, (_layer)->pld_len); \
} }
@@ -865,7 +864,7 @@ static inline const char *parse_ipv4(struct packet *handler, const char *data, u
// ip fragmented // ip fragmented
if ((ntohs(hdr->ip_off) & IP_MF) || (ntohs(hdr->ip_off) & IP_OFFMASK)) if ((ntohs(hdr->ip_off) & IP_MF) || (ntohs(hdr->ip_off) & IP_OFFMASK))
{ {
PACKET_LOG_DEBUG("%s: ip is fragmented", LOG_PACKET); PACKET_LOG_DEBUG("ip is fragmented");
return layer->pld_ptr; return layer->pld_ptr;
} }
@@ -1471,14 +1470,5 @@ uint64_t packet_get_hash(const struct packet *handler, enum ldbc_method method,
return hash_value; return hash_value;
} }
#if 0
char *inner_addr_str = tuple2_tostring(&inner_addr);
char *outer_addr_str = tuple2_tostring(&outer_addr);
printf("%s: outer_addr: %s, inner_addr: %s, dir: %s, hash_method: %s, hash_value: %lu\n",
LOG_PACKET, outer_addr_str, inner_addr_str, (direction ? "E2I" : "I2E"), ldbc_method_tostring(method), hash_value);
free(inner_addr_str);
free(outer_addr_str);
#endif
return hash_value; return hash_value;
} }

4
src/packet/packet.h
View File

@@ -14,12 +14,12 @@ extern "C"
// #define PACKET_LOG_ERROR(format, ...) void(0) // #define PACKET_LOG_ERROR(format, ...) void(0)
#ifndef PACKET_LOG_ERROR #ifndef PACKET_LOG_ERROR
#define PACKET_LOG_ERROR(format, ...) \ #define PACKET_LOG_ERROR(format, ...) \
fprintf(stderr, "ERROR " format "\n", ##__VA_ARGS__); fprintf(stderr, "ERROR (packet), " format "\n", ##__VA_ARGS__);
#endif #endif
// #define PACKET_LOG_DEBUG(format, ...) void(0) // #define PACKET_LOG_DEBUG(format, ...) void(0)
#ifndef PACKET_LOG_DEBUG #ifndef PACKET_LOG_DEBUG
#define PACKET_LOG_DEBUG(format, ...) \ #define PACKET_LOG_DEBUG(format, ...) \
fprintf(stderr, "DEBUG " format "\n", ##__VA_ARGS__); fprintf(stderr, "DEBUG (packet), " format "\n", ##__VA_ARGS__);
#endif #endif
enum layer_type enum layer_type

19
src/session/CMakeLists.txt
View File

@@ -7,15 +7,16 @@ add_library(session_manager
session_pool.cpp session_pool.cpp
session_table.cpp session_table.cpp
session_timer.cpp session_timer.cpp
#session_manager.cpp session_queue.cpp
) session_manager.cpp
)
target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/uthash) target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/uthash)
target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/timeout) target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/timeout)
target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/src/packet) target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/src/packet)
target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/src/session) target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/src/session)
target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/src/timestamp) target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/src/timestamp)
target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/src/tuple) target_include_directories(session_manager PUBLIC ${CMAKE_SOURCE_DIR}/src/tuple)
target_link_libraries(session_manager timeout timestamp tuple) target_link_libraries(session_manager timeout timestamp tuple packet)
############################################################################### ###############################################################################
# gtest # gtest
@@ -37,8 +38,18 @@ add_executable(gtest_session_timer gtest_session_timer.cpp)
target_include_directories(gtest_session_timer PUBLIC ${CMAKE_CURRENT_LIST_DIR}) target_include_directories(gtest_session_timer PUBLIC ${CMAKE_CURRENT_LIST_DIR})
target_link_libraries(gtest_session_timer session_manager gtest) target_link_libraries(gtest_session_timer session_manager gtest)
add_executable(gtest_session_queue gtest_session_queue.cpp)
target_include_directories(gtest_session_queue PUBLIC ${CMAKE_CURRENT_LIST_DIR})
target_link_libraries(gtest_session_queue session_manager gtest)
add_executable(gtest_session_manager gtest_session_manager.cpp)
target_include_directories(gtest_session_manager PUBLIC ${CMAKE_CURRENT_LIST_DIR})
target_link_libraries(gtest_session_manager session_manager gtest)
include(GoogleTest) include(GoogleTest)
gtest_discover_tests(gtest_session) gtest_discover_tests(gtest_session)
gtest_discover_tests(gtest_session_pool) gtest_discover_tests(gtest_session_pool)
gtest_discover_tests(gtest_session_table) gtest_discover_tests(gtest_session_table)
gtest_discover_tests(gtest_session_timer) gtest_discover_tests(gtest_session_timer)
gtest_discover_tests(gtest_session_queue)
gtest_discover_tests(gtest_session_manager)

1198
src/session/gtest_session_manager.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

29
src/session/gtest_session_queue.cpp Normal file
View File

@@ -0,0 +1,29 @@
#include <gtest/gtest.h>
#include "session_private.h"
#include "session_queue.h"
TEST(SESSION_QUEUE, POP_PUSH)
{
struct session sess1;
struct session sess2;
struct session_queue *queue = NULL;
queue = session_queue_create();
EXPECT_TRUE(queue != NULL);
EXPECT_TRUE(session_queue_pop(queue) == NULL);
session_queue_push(queue, &sess1);
session_queue_push(queue, &sess2);
EXPECT_TRUE(session_queue_pop(queue) == &sess1);
EXPECT_TRUE(session_queue_pop(queue) == &sess2);
EXPECT_TRUE(session_queue_pop(queue) == NULL);
session_queue_destroy(queue);
}
int main(int argc, char **argv)
{
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

28
src/session/gtest_session_table.cpp
View File

@@ -147,46 +147,46 @@ TEST(SESSION_TABLE, FIND_OLDEST_NEWEST)
// Add Session // Add Session
EXPECT_TRUE(session_table_find_oldest_session(sess_table) == NULL); EXPECT_TRUE(session_table_find_least_recently_unused_session(sess_table) == NULL);
EXPECT_TRUE(session_table_find_newest_session(sess_table) == NULL); EXPECT_TRUE(session_table_find_least_recently_used_session(sess_table) == NULL);
sess1 = session_pool_alloc(sess_pool); sess1 = session_pool_alloc(sess_pool);
EXPECT_TRUE(sess1 != NULL); EXPECT_TRUE(sess1 != NULL);
session_set_id(sess1, 1); session_set_id(sess1, 1);
session_set_tuple6(sess1, &tuple_1); session_set_tuple6(sess1, &tuple_1);
EXPECT_TRUE(session_table_add_session(sess_table, &tuple_1, sess1) == 0); EXPECT_TRUE(session_table_add_session(sess_table, &tuple_1, sess1) == 0);
EXPECT_TRUE(session_table_find_oldest_session(sess_table) == sess1); EXPECT_TRUE(session_table_find_least_recently_unused_session(sess_table) == sess1);
EXPECT_TRUE(session_table_find_newest_session(sess_table) == sess1); EXPECT_TRUE(session_table_find_least_recently_used_session(sess_table) == sess1);
sess2 = session_pool_alloc(sess_pool); sess2 = session_pool_alloc(sess_pool);
EXPECT_TRUE(sess2 != NULL); EXPECT_TRUE(sess2 != NULL);
session_set_id(sess2, 2); session_set_id(sess2, 2);
session_set_tuple6(sess2, &tuple_2); session_set_tuple6(sess2, &tuple_2);
EXPECT_TRUE(session_table_add_session(sess_table, &tuple_2, sess2) == 0); EXPECT_TRUE(session_table_add_session(sess_table, &tuple_2, sess2) == 0);
EXPECT_TRUE(session_table_find_oldest_session(sess_table) == sess1); EXPECT_TRUE(session_table_find_least_recently_unused_session(sess_table) == sess1);
EXPECT_TRUE(session_table_find_newest_session(sess_table) == sess2); EXPECT_TRUE(session_table_find_least_recently_used_session(sess_table) == sess2);
sess3 = session_pool_alloc(sess_pool); sess3 = session_pool_alloc(sess_pool);
EXPECT_TRUE(sess3 != NULL); EXPECT_TRUE(sess3 != NULL);
session_set_id(sess3, 3); session_set_id(sess3, 3);
session_set_tuple6(sess3, &tuple_3); session_set_tuple6(sess3, &tuple_3);
EXPECT_TRUE(session_table_add_session(sess_table, &tuple_3, sess3) == 0); EXPECT_TRUE(session_table_add_session(sess_table, &tuple_3, sess3) == 0);
EXPECT_TRUE(session_table_find_oldest_session(sess_table) == sess1); EXPECT_TRUE(session_table_find_least_recently_unused_session(sess_table) == sess1);
EXPECT_TRUE(session_table_find_newest_session(sess_table) == sess3); EXPECT_TRUE(session_table_find_least_recently_used_session(sess_table) == sess3);
// Delete Session // Delete Session
session_table_delete_session(sess_table, &tuple_1); session_table_delete_session(sess_table, &tuple_1);
EXPECT_TRUE(session_table_find_oldest_session(sess_table) == sess2); EXPECT_TRUE(session_table_find_least_recently_unused_session(sess_table) == sess2);
EXPECT_TRUE(session_table_find_newest_session(sess_table) == sess3); EXPECT_TRUE(session_table_find_least_recently_used_session(sess_table) == sess3);
session_table_delete_session(sess_table, &tuple_2); session_table_delete_session(sess_table, &tuple_2);
EXPECT_TRUE(session_table_find_oldest_session(sess_table) == sess3); EXPECT_TRUE(session_table_find_least_recently_unused_session(sess_table) == sess3);
EXPECT_TRUE(session_table_find_newest_session(sess_table) == sess3); EXPECT_TRUE(session_table_find_least_recently_used_session(sess_table) == sess3);
session_table_delete_session(sess_table, &tuple_3); session_table_delete_session(sess_table, &tuple_3);
EXPECT_TRUE(session_table_find_oldest_session(sess_table) == NULL); EXPECT_TRUE(session_table_find_least_recently_unused_session(sess_table) == NULL);
EXPECT_TRUE(session_table_find_newest_session(sess_table) == NULL); EXPECT_TRUE(session_table_find_least_recently_used_session(sess_table) == NULL);
// Destroy // Destroy
session_table_destroy(sess_table); session_table_destroy(sess_table);

231
src/session/session.cpp
View File

@@ -18,6 +18,8 @@ struct ex_manager
}; };
static struct ex_manager g_ex_manager = {0}; static struct ex_manager g_ex_manager = {0};
uint8_t tcp_builtin_ex = 0; // built-in ex_data index
uint8_t udp_builtin_ex = 0; // built-in ex_data index
/****************************************************************************** /******************************************************************************
* ev queue * ev queue
@@ -81,7 +83,7 @@ void session_set_id(struct session *sess, uint64_t id)
sess->id = id; sess->id = id;
} }
uint64_t session_get_id(struct session *sess) uint64_t session_get_id(const struct session *sess)
{ {
return sess->id; return sess->id;
} }
@@ -92,18 +94,28 @@ void session_set_tuple6(struct session *sess, struct tuple6 *tuple)
memcpy(&sess->tuple, tuple, sizeof(struct tuple6)); memcpy(&sess->tuple, tuple, sizeof(struct tuple6));
} }
struct tuple6 *session_get0_tuple6(struct session *sess) const struct tuple6 *session_get0_tuple6(const struct session *sess)
{ {
return &sess->tuple; return &sess->tuple;
} }
void session_set_tuple6_dir(struct session *sess, enum session_dir dir)
{
sess->tuple_dir = dir;
}
enum session_dir session_get_tuple6_dir(const struct session *sess)
{
return sess->tuple_dir;
}
// session state // session state
void session_set_state(struct session *sess, enum session_state state) void session_set_state(struct session *sess, enum session_state state)
{ {
sess->state = state; sess->state = state;
} }
enum session_state session_get_state(struct session *sess) enum session_state session_get_state(const struct session *sess)
{ {
return sess->state; return sess->state;
} }
@@ -114,7 +126,7 @@ void session_set_type(struct session *sess, enum session_type type)
sess->type = type; sess->type = type;
} }
enum session_type session_get_type(struct session *sess) enum session_type session_get_type(const struct session *sess)
{ {
return sess->type; return sess->type;
} }
@@ -132,22 +144,22 @@ void session_inc_s2c_metrics(struct session *sess, uint64_t packets, uint64_t by
sess->s2c_packets += packets; sess->s2c_packets += packets;
} }
uint64_t session_get_c2s_bytes(struct session *sess) uint64_t session_get_c2s_bytes(const struct session *sess)
{ {
return sess->c2s_bytes; return sess->c2s_bytes;
} }
uint64_t session_get_s2c_bytes(struct session *sess) uint64_t session_get_s2c_bytes(const struct session *sess)
{ {
return sess->s2c_bytes; return sess->s2c_bytes;
} }
uint64_t session_get_c2s_packets(struct session *sess) uint64_t session_get_c2s_packets(const struct session *sess)
{ {
return sess->c2s_packets; return sess->c2s_packets;
} }
uint64_t session_get_s2c_packets(struct session *sess) uint64_t session_get_s2c_packets(const struct session *sess)
{ {
return sess->s2c_packets; return sess->s2c_packets;
} }
@@ -163,7 +175,7 @@ void session_set_s2c_1st_md(struct session *sess, struct metadata *md)
memcpy(&sess->s2c_1st_md, md, sizeof(struct metadata)); memcpy(&sess->s2c_1st_md, md, sizeof(struct metadata));
} }
struct metadata *session_get0_c2s_1st_md(struct session *sess) const struct metadata *session_get0_c2s_1st_md(const struct session *sess)
{ {
if (sess->c2s_1st_md.len == 0) if (sess->c2s_1st_md.len == 0)
{ {
@@ -172,7 +184,7 @@ struct metadata *session_get0_c2s_1st_md(struct session *sess)
return &sess->c2s_1st_md; return &sess->c2s_1st_md;
} }
struct metadata *session_get0_s2c_1st_md(struct session *sess) const struct metadata *session_get0_s2c_1st_md(const struct session *sess)
{ {
if (sess->s2c_1st_md.len == 0) if (sess->s2c_1st_md.len == 0)
{ {
@@ -192,12 +204,12 @@ void session_set_last_time(struct session *sess, uint64_t timestamp)
sess->last_time = timestamp; sess->last_time = timestamp;
} }
uint64_t session_get_create_time(struct session *sess) uint64_t session_get_create_time(const struct session *sess)
{ {
return sess->create_time; return sess->create_time;
} }
uint64_t session_get_last_time(struct session *sess) uint64_t session_get_last_time(const struct session *sess)
{ {
return sess->last_time; return sess->last_time;
} }
@@ -212,7 +224,7 @@ void session_set0_cur_pkt(struct session *sess, const struct packet *pkt)
sess->cur_pkt = pkt; sess->cur_pkt = pkt;
} }
const struct packet *session_get0_cur_pkt(struct session *sess) const struct packet *session_get0_cur_pkt(const struct session *sess)
{ {
return sess->cur_pkt; return sess->cur_pkt;
} }
@@ -223,7 +235,7 @@ void session_set_cur_dir(struct session *sess, enum session_dir dir)
sess->cur_dir = dir; sess->cur_dir = dir;
} }
enum session_dir session_get_cur_dir(struct session *sess) enum session_dir session_get_cur_dir(const struct session *sess)
{ {
return sess->cur_dir; return sess->cur_dir;
} }
@@ -301,7 +313,7 @@ void session_set_ex_data(struct session *sess, uint8_t idx, void *val)
sess->ex_data[idx] = val; sess->ex_data[idx] = val;
} }
void *session_get0_ex_data(struct session *sess, uint8_t idx) void *session_get0_ex_data(const struct session *sess, uint8_t idx)
{ {
if (idx >= g_ex_manager.count) if (idx >= g_ex_manager.count)
{ {
@@ -365,4 +377,193 @@ void session_run_expirecb(struct session *sess)
{ {
sess->expire_cb(sess, sess->expire_arg); sess->expire_cb(sess, sess->expire_arg);
} }
}
/******************************************************************************
* session dump
******************************************************************************/
static void tcp_ex_data_tostring(uint64_t ex_data, char *buffer, size_t buffer_len)
{
if (ex_data == 0)
{
return;
}
int nused = 0;
if (ex_data & TCP_SYN_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "TCP_SYN_RECVED ");
}
if (ex_data & TCP_SYNACK_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "TCP_SYNACK_RECVED ");
}
if (ex_data & TCP_ACK_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "TCP_ACK_RECVED ");
}
if (ex_data & TCP_C2S_PAYLOAD_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "TCP_C2S_PAYLOAD_RECVED ");
}
if (ex_data & TCP_S2C_PAYLOAD_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "TCP_S2C_PAYLOAD_RECVED ");
}
if (ex_data & TCP_C2S_FIN_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "TCP_C2S_FIN_RECVED ");
}
if (ex_data & TCP_S2C_FIN_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "TCP_S2C_FIN_RECVED ");
}
if (ex_data & TCP_C2S_RST_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "TCP_C2S_RST_RECVED ");
}
if (ex_data & TCP_S2C_RST_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "TCP_S2C_RST_RECVED ");
}
}
static void udp_ex_data_tostring(uint64_t ex_data, char *buffer, size_t buffer_len)
{
if (ex_data == 0)
{
return;
}
int nused = 0;
if (ex_data & UDP_C2S_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "UDP_C2S_RECVED ");
}
if (ex_data & UDP_S2C_RECVED)
{
snprintf(buffer + nused, buffer_len - nused, "UDP_S2C_RECVED ");
}
}
const char *session_event_tostring(enum session_event event)
{
switch (event)
{
case SESSION_EVENT_NONE:
return "none";
case SESSION_EVENT_OPENING:
return "opening";
case SESSION_EVENT_ACTIVE:
return "active";
case SESSION_EVENT_CLOSING:
return "closing";
default:
return "unknown";
}
}
const char *session_state_tostring(enum session_state state)
{
switch (state)
{
case SESSION_STATE_INIT:
return "init";
case SESSION_STATE_OPENING:
return "opening";
case SESSION_STATE_ACTIVE:
return "active";
case SESSION_STATE_DISCARD:
return "discard";
case SESSION_STATE_CLOSING:
return "closing";
case SESSION_STATE_CLOSED:
return "closed";
default:
return "unknown";
}
}
const char *session_type_tostring(enum session_type type)
{
switch (type)
{
case SESSION_TYPE_NONE:
return "none";
case SESSION_TYPE_TCP:
return "tcp";
case SESSION_TYPE_TCP_STREAM:
return "tcp_stream";
case SESSION_TYPE_UDP:
return "udp";
default:
return "unknown";
}
}
const char *session_dir_tostring(enum session_dir dir)
{
switch (dir)
{
case SESSION_DIR_NONE:
return "none";
case SESSION_DIR_C2S:
return "c2s";
case SESSION_DIR_S2C:
return "s2c";
default:
return "unknown";
}
}
void session_dump(struct session *sess)
{
char buffer[128] = {0};
tuple6_tostring(session_get0_tuple6(sess), buffer, sizeof(buffer));
printf("session id : %" PRIu64 "\n", session_get_id(sess));
printf("session tuple6 : %s\n", buffer);
printf("session tuple6 dir : %s\n", session_dir_tostring(session_get_tuple6_dir(sess)));
printf("session state : %s\n", session_state_tostring(session_get_state(sess)));
printf("session type : %s\n", session_type_tostring(session_get_type(sess)));
printf("session c2s packets : %" PRIu64 "\n", session_get_c2s_packets(sess));
printf("session c2s bytes : %" PRIu64 "\n", session_get_c2s_bytes(sess));
printf("session s2c packets : %" PRIu64 "\n", session_get_s2c_packets(sess));
printf("session s2c bytes : %" PRIu64 "\n", session_get_s2c_bytes(sess));
printf("session c2s 1st metadata : %p\n", (void *)session_get0_c2s_1st_md(sess));
printf("session s2c 1st metadata : %p\n", (void *)session_get0_s2c_1st_md(sess));
printf("session create time : %" PRIu64 "\n", session_get_create_time(sess));
printf("session last time : %" PRIu64 "\n", session_get_last_time(sess));
printf("session current packet : %p\n", (void *)session_get0_cur_pkt(sess));
printf("session current dir : %s\n", session_dir_tostring(session_get_cur_dir(sess)));
printf("session ex data: \n");
for (uint8_t i = 0; i < g_ex_manager.count; i++)
{
if (i == tcp_builtin_ex)
{
memset(buffer, 0, sizeof(buffer));
tcp_ex_data_tostring((uint64_t)sess->ex_data[i], buffer, sizeof(buffer));
printf(" ex_idx: %d, ex_key: %s, ex_val: %s\n", i, g_ex_manager.schemas[i].key, buffer);
}
else if (i == udp_builtin_ex)
{
memset(buffer, 0, sizeof(buffer));
udp_ex_data_tostring((uint64_t)sess->ex_data[i], buffer, sizeof(buffer));
printf(" ex_idx: %d, ex_key: %s, ex_val: %s\n", i, g_ex_manager.schemas[i].key, buffer);
}
else
{
printf(" ex_idx: %d, ex_key: %s, ex_val: %p\n", i, g_ex_manager.schemas[i].key, sess->ex_data[i]);
}
}
} }

48
src/session/session.h
View File

@@ -9,6 +9,7 @@ extern "C"
#include <stdint.h> #include <stdint.h>
#include "tuple.h" #include "tuple.h"
#include "packet.h"
enum session_state enum session_state
{ {
@@ -42,8 +43,9 @@ enum session_event
enum session_dir enum session_dir
{ {
SESSION_DIR_C2S = 0, SESSION_DIR_NONE = 0,
SESSION_DIR_S2C = 1, SESSION_DIR_C2S = 1,
SESSION_DIR_S2C = 2,
}; };
struct metadata struct metadata
@@ -62,39 +64,41 @@ void session_init(struct session *sess);
// session id // session id
void session_set_id(struct session *sess, uint64_t id); void session_set_id(struct session *sess, uint64_t id);
uint64_t session_get_id(struct session *sess); uint64_t session_get_id(const struct session *sess);
// session key // session key
void session_set_tuple6(struct session *sess, struct tuple6 *tuple); void session_set_tuple6(struct session *sess, struct tuple6 *tuple);
struct tuple6 *session_get0_tuple6(struct session *sess); const struct tuple6 *session_get0_tuple6(const struct session *sess);
void session_set_tuple6_dir(struct session *sess, enum session_dir dir);
enum session_dir session_get_tuple6_dir(const struct session *sess);
// session state // session state
void session_set_state(struct session *sess, enum session_state state); void session_set_state(struct session *sess, enum session_state state);
enum session_state session_get_state(struct session *sess); enum session_state session_get_state(const struct session *sess);
// session type // session type
void session_set_type(struct session *sess, enum session_type type); void session_set_type(struct session *sess, enum session_type type);
enum session_type session_get_type(struct session *sess); enum session_type session_get_type(const struct session *sess);
// session metrics // session metrics
void session_inc_c2s_metrics(struct session *sess, uint64_t packets, uint64_t bytes); void session_inc_c2s_metrics(struct session *sess, uint64_t packets, uint64_t bytes);
void session_inc_s2c_metrics(struct session *sess, uint64_t packets, uint64_t bytes); void session_inc_s2c_metrics(struct session *sess, uint64_t packets, uint64_t bytes);
uint64_t session_get_c2s_bytes(struct session *sess); uint64_t session_get_c2s_bytes(const struct session *sess);
uint64_t session_get_s2c_bytes(struct session *sess); uint64_t session_get_s2c_bytes(const struct session *sess);
uint64_t session_get_c2s_packets(struct session *sess); uint64_t session_get_c2s_packets(const struct session *sess);
uint64_t session_get_s2c_packets(struct session *sess); uint64_t session_get_s2c_packets(const struct session *sess);
// session metadata // session metadata
void session_set_c2s_1st_md(struct session *sess, struct metadata *md); void session_set_c2s_1st_md(struct session *sess, struct metadata *md);
void session_set_s2c_1st_md(struct session *sess, struct metadata *md); void session_set_s2c_1st_md(struct session *sess, struct metadata *md);
struct metadata *session_get0_c2s_1st_md(struct session *sess); const struct metadata *session_get0_c2s_1st_md(const struct session *sess);
struct metadata *session_get0_s2c_1st_md(struct session *sess); const struct metadata *session_get0_s2c_1st_md(const struct session *sess);
// session timestamp // session timestamp
void session_set_create_time(struct session *sess, uint64_t timestamp); void session_set_create_time(struct session *sess, uint64_t timestamp);
void session_set_last_time(struct session *sess, uint64_t timestamp); void session_set_last_time(struct session *sess, uint64_t timestamp);
uint64_t session_get_create_time(struct session *sess); uint64_t session_get_create_time(const struct session *sess);
uint64_t session_get_last_time(struct session *sess); uint64_t session_get_last_time(const struct session *sess);
/****************************************************************************** /******************************************************************************
* session current packet * session current packet
@@ -102,11 +106,11 @@ uint64_t session_get_last_time(struct session *sess);
// session current packet // session current packet
void session_set0_cur_pkt(struct session *sess, const struct packet *pkt); void session_set0_cur_pkt(struct session *sess, const struct packet *pkt);
const struct packet *session_get0_cur_pkt(struct session *sess); const struct packet *session_get0_cur_pkt(const struct session *sess);
// session current dir // session current dir
void session_set_cur_dir(struct session *sess, enum session_dir dir); void session_set_cur_dir(struct session *sess, enum session_dir dir);
enum session_dir session_get_cur_dir(struct session *sess); enum session_dir session_get_cur_dir(const struct session *sess);
/****************************************************************************** /******************************************************************************
* session event * session event
@@ -139,7 +143,7 @@ uint8_t session_get_ex_new_index(const char *key, session_ex_free_cb *free_cb, v
* if key not exist: set new value. * if key not exist: set new value.
*/ */
void session_set_ex_data(struct session *sess, uint8_t idx, void *val); void session_set_ex_data(struct session *sess, uint8_t idx, void *val);
void *session_get0_ex_data(struct session *sess, uint8_t idx); void *session_get0_ex_data(const struct session *sess, uint8_t idx);
/* /*
* after set ex_data, the owner of ex_data is session, so user should not free it directly. * after set ex_data, the owner of ex_data is session, so user should not free it directly.
* if user want to free ex_data, should use session_free_ex_data. * if user want to free ex_data, should use session_free_ex_data.
@@ -157,6 +161,16 @@ void session_set_expirecb(struct session *sess, session_expire_cb expire_cb, voi
void session_del_expirecb(struct session *sess); void session_del_expirecb(struct session *sess);
void session_run_expirecb(struct session *sess); void session_run_expirecb(struct session *sess);
/******************************************************************************
* session dump
******************************************************************************/
const char *session_event_tostring(enum session_event event);
const char *session_state_tostring(enum session_state state);
const char *session_type_tostring(enum session_type type);
const char *session_dir_tostring(enum session_dir dir);
void session_dump(struct session *sess);
#ifdef __cpluscplus #ifdef __cpluscplus
} }
#endif #endif

649
src/session/session_manager.cpp
View File

@@ -6,110 +6,507 @@
#include "session_pool.h" #include "session_pool.h"
#include "session_table.h" #include "session_table.h"
#include "session_timer.h" #include "session_timer.h"
#include "session_queue.h"
#include "session_private.h" #include "session_private.h"
#include "packet_helpers.h"
struct session_manager struct session_manager
{ {
struct session_pool *sess_pool; struct session_pool *sess_pool;
struct session_table *sess_table; struct session_table *sess_table;
struct session_timer *sess_timer; struct session_timer *sess_timer;
struct session_queue *sess_queue; // ready session queue
session_event_cb event_cb; session_event_cb event_cb;
void *arg; void *arg;
struct session *queue_head_ptr; // timeout config
struct session *queue_tail_ptr; uint64_t packet_timeout_ms;
uint64_t closing_timeout_ms;
// 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;
}; };
#define HANDSHAKE_TIMEOUT_MS (5 * 1000)
#define DATA_TIMEOUT_MS (60 * 1000)
/****************************************************************************** /******************************************************************************
* private API * utils
******************************************************************************/ ******************************************************************************/
static void session_manager_handle_new_session(struct session_manager *mgr, struct session_key *key, struct session *sess, const struct packet *pkt) // TODO
static uint64_t alloc_session_id(void)
{ {
// TODO return 0;
} }
static void session_manager_handle_old_session(struct session_manager *mgr, struct session_key *key, struct session *sess, const struct packet *pkt) // TODO
struct metadata *packet_get0_metadata(const struct packet *pkt)
{ {
// TODO static struct metadata md = {0};
return &md;
};
/******************************************************************************
* 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++;
}
}
} }
// Enqueue ready session to queue tail static void update_counter_on_active(struct session_manager *mgr, struct session *sess)
static void session_manager_enqueue_ready_session(struct session_manager *mgr, struct session *sess)
{ {
if (mgr == NULL || sess == NULL) 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 event and 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);
session_set_expirecb(sess, cb, mgr, timestamp_get_msec() + timeout_ms);
session_timer_add_session(mgr->sess_timer, sess);
}
/******************************************************************************
* expire callback
******************************************************************************/
static void closing_expire_callback(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 packet_expire_callback(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_manager_trigger_session_event(mgr, sess, SESSION_EVENT_CLOSING);
session_manager_update_session_timer(mgr, sess, closing_expire_callback, mgr->closing_timeout_ms);
}
/******************************************************************************
* 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);
struct metadata *md = packet_get0_metadata(pkt);
if (curr_dir == SESSION_DIR_C2S)
{
session_inc_c2s_metrics(sess, 1, len);
if (session_get0_c2s_1st_md(sess) == NULL)
{
session_set_c2s_1st_md(sess, md);
}
}
else
{
session_inc_s2c_metrics(sess, 1, len);
if (session_get0_s2c_1st_md(sess) == NULL)
{
session_set_s2c_1st_md(sess, md);
}
}
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)
{
uint64_t state = (uint64_t)session_get0_ex_data(sess, tcp_builtin_ex);
if (packet_has_tcp_flag_rst(pkt))
{
if (curr_dir == SESSION_DIR_C2S)
{
session_set_ex_data(sess, tcp_builtin_ex, (void *)(state & TCP_C2S_RST_RECVED));
}
else
{
session_set_ex_data(sess, tcp_builtin_ex, (void *)(state & TCP_S2C_RST_RECVED));
}
}
if (packet_has_tcp_flag_fin(pkt))
{
if (curr_dir == SESSION_DIR_C2S)
{
session_set_ex_data(sess, tcp_builtin_ex, (void *)(state & TCP_C2S_FIN_RECVED));
}
else
{
session_set_ex_data(sess, tcp_builtin_ex, (void *)(state & TCP_S2C_FIN_RECVED));
}
}
if (packet_has_tcp_flag_syn(pkt))
{
if (packet_has_tcp_flag_ack(pkt))
{
session_set_ex_data(sess, tcp_builtin_ex, (void *)(state | TCP_SYNACK_RECVED));
}
else
{
session_set_ex_data(sess, tcp_builtin_ex, (void *)(state | TCP_SYN_RECVED));
}
}
if (packet_has_tcp_flag_ack(pkt) && curr_dir == SESSION_DIR_C2S)
{
session_set_ex_data(sess, tcp_builtin_ex, (void *)(state | TCP_ACK_RECVED));
}
if (packet_get_tcp_pld_len(pkt) > 0)
{
if (curr_dir == SESSION_DIR_C2S)
{
session_set_ex_data(sess, tcp_builtin_ex, (void *)(state | TCP_C2S_PAYLOAD_RECVED));
}
else
{
session_set_ex_data(sess, tcp_builtin_ex, (void *)(state | TCP_S2C_PAYLOAD_RECVED));
}
}
}
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)
{
if (!packet_has_tcp_flag_syn(pkt))
{
// not syn packet, discard
return -1;
}
enum session_dir curr_dir = SESSION_DIR_NONE;
session_init(sess);
// syn packet
if (!packet_has_tcp_flag_ack(pkt))
{
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_manager_trigger_session_event(mgr, sess, SESSION_EVENT_OPENING);
session_manager_update_session_timer(mgr, sess, packet_expire_callback, mgr->packet_timeout_ms);
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_manager_trigger_session_event(mgr, sess, SESSION_EVENT_OPENING);
session_manager_trigger_session_event(mgr, sess, SESSION_EVENT_ACTIVE);
session_manager_update_session_timer(mgr, sess, packet_expire_callback, mgr->packet_timeout_ms);
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_manager_trigger_session_event(mgr, sess, SESSION_EVENT_CLOSING);
session_manager_update_session_timer(mgr, sess, closing_expire_callback, mgr->closing_timeout_ms);
return;
}
if (tcp_need_active(state))
{
update_counter_on_active(mgr, sess);
session_set_state(sess, SESSION_STATE_ACTIVE);
session_manager_trigger_session_event(mgr, sess, SESSION_EVENT_ACTIVE);
session_manager_update_session_timer(mgr, sess, packet_expire_callback, mgr->packet_timeout_ms);
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_manager_trigger_session_event(mgr, sess, SESSION_EVENT_ACTIVE);
session_manager_update_session_timer(mgr, sess, packet_expire_callback, mgr->packet_timeout_ms);
}
// 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 (packet_has_tcp(pkt))
{
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; return;
} }
if (mgr->queue_head_ptr == NULL) if (packet_has_tcp(pkt))
{ {
mgr->queue_head_ptr = sess; handle_tcp_old_session(mgr, key, sess, pkt);
mgr->queue_tail_ptr = sess;
sess->next_ready_ptr = NULL;
} }
else else
{ {
mgr->queue_tail_ptr->next_ready_ptr = sess; handle_udp_old_session(mgr, key, sess, pkt);
mgr->queue_tail_ptr = sess;
sess->next_ready_ptr = NULL;
} }
} }
// Dequeue ready session from queue head
struct session *session_manager_ready_session_dequeue(struct session_manager *mgr)
{
if (mgr == NULL)
{
return NULL;
}
struct session *sess = mgr->queue_head_ptr;
if (sess == NULL)
{
return NULL;
}
if (mgr->queue_head_ptr == mgr->queue_tail_ptr)
{
mgr->queue_head_ptr = NULL;
mgr->queue_tail_ptr = NULL;
}
else
{
mgr->queue_head_ptr = sess->next_ready_ptr;
}
sess->next_ready_ptr = NULL;
return sess;
}
// handshake expire set to discard
static void handshake_expire_cb(struct session *sess, void *arg)
{
struct session_manager *mgr = (struct session_manager *)arg;
assert(mgr != NULL);
session_set_state(sess, SESSION_STATE_DISCARD);
session_push_event(sess, SESSION_EVENT_CLOSING);
session_manager_enqueue_ready_session(mgr, sess);
}
// data expire set to closing
static void data_expire_cb(struct session *sess, void *arg)
{
struct session_manager *mgr = (struct session_manager *)arg;
assert(mgr != NULL);
session_set_state(sess, SESSION_STATE_CLOSING);
session_push_event(sess, SESSION_EVENT_CLOSING);
session_manager_enqueue_ready_session(mgr, sess);
}
/****************************************************************************** /******************************************************************************
* public API * public API
******************************************************************************/ ******************************************************************************/
@@ -140,8 +537,25 @@ struct session_manager *session_manager_create(uint64_t max_session_num)
goto error; goto error;
} }
mgr->queue_head_ptr = NULL; mgr->sess_queue = session_queue_create();
mgr->queue_tail_ptr = NULL; if (mgr->sess_queue == NULL)
{
goto error;
}
mgr->closing_timeout_ms = 2 * 1000;
mgr->packet_timeout_ms = 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);
return mgr; return mgr;
@@ -154,6 +568,7 @@ void session_manager_destroy(struct session_manager *mgr)
{ {
if (mgr) if (mgr)
{ {
session_queue_destroy(mgr->sess_queue);
session_timer_destroy(mgr->sess_timer); session_timer_destroy(mgr->sess_timer);
session_table_destroy(mgr->sess_table); session_table_destroy(mgr->sess_table);
session_pool_destroy(mgr->sess_pool); session_pool_destroy(mgr->sess_pool);
@@ -168,31 +583,60 @@ void session_manager_set_session_eventcb(struct session_manager *mgr, session_ev
mgr->arg = arg; mgr->arg = arg;
} }
void session_manager_set_packet_timeout(struct session_manager *mgr, uint64_t timeout_ms)
{
mgr->packet_timeout_ms = timeout_ms;
}
void session_manager_set_closing_timeout(struct session_manager *mgr, uint64_t timeout_ms)
{
mgr->closing_timeout_ms = 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_find_session(struct session_manager *mgr, const struct packet *pkt)
{ {
struct session_key key; struct tuple6 key;
// TODO packet to key if (packet_get_innermost_tuple6(pkt, &key) == -1)
{
return NULL;
}
struct session *sess = session_table_find_session(mgr->sess_table, &key); struct session *sess = session_table_find_session(mgr->sess_table, &key);
if (sess == NULL) if (sess == NULL)
{ {
// if session pool is full, discard oldest session
if (session_pool_get_count(mgr->sess_pool) == 1) if (session_pool_get_count(mgr->sess_pool) == 1)
{ {
struct session *oldest_sess = session_table_find_oldest_session(mgr->sess_table); struct session *unused_sess = session_table_find_least_recently_unused_session(mgr->sess_table);
assert(oldest_sess == NULL); assert(unused_sess);
session_set_state(oldest_sess, SESSION_STATE_DISCARD);
session_push_event(oldest_sess, SESSION_EVENT_CLOSING); update_counter_on_closing(mgr, unused_sess);
session_manager_enqueue_ready_session(mgr, oldest_sess); session_set_state(unused_sess, SESSION_STATE_DISCARD);
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);
} }
sess = session_pool_alloc(mgr->sess_pool); sess = session_pool_alloc(mgr->sess_pool);
assert(sess != NULL); assert(sess != NULL);
session_manager_handle_new_session(mgr, &key, sess, pkt); // 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 else
{ {
session_manager_handle_old_session(mgr, &key, sess, pkt); handle_old_session(mgr, &key, sess, pkt);
} }
return sess; return sess;
@@ -206,19 +650,24 @@ void session_manager_dispatch(struct session_manager *mgr)
void *cb_arg = mgr->arg; void *cb_arg = mgr->arg;
session_event_cb event_cb = mgr->event_cb; session_event_cb event_cb = mgr->event_cb;
while (1) SESSION_MANAGER_LOG_DEBUG("current timestamp: %lu s", timestamp_get_sec());
// limit expire session number
for (int i = 0; i < 100; i++)
{ {
sess = session_timer_expire_session(mgr->sess_timer, timestamp_get_msec()); sess = session_timer_expire_session(mgr->sess_timer, timestamp_get_msec());
if (sess == NULL) if (sess == NULL)
{ {
break; break;
} }
session_run_expirecb(sess); session_run_expirecb(sess);
} }
while (1) while (1)
{ {
sess = session_manager_ready_session_dequeue(mgr); // get session from ready queue
sess = session_queue_pop(mgr->sess_queue);
if (sess == NULL) if (sess == NULL)
{ {
break; break;
@@ -226,15 +675,49 @@ void session_manager_dispatch(struct session_manager *mgr)
while (1) while (1)
{ {
// get event from session
if (session_pop_event(sess, &event) == false) if (session_pop_event(sess, &event) == false)
{ {
break; break;
} }
SESSION_MANAGER_LOG_DEBUG("handle \"%s\" event on session %lu", session_event_tostring((enum session_event)event), session_get_id(sess));
if (event_cb) if (event_cb)
{ {
event_cb(sess, event, cb_arg); event_cb(sess, event, cb_arg);
} }
} }
session_set0_cur_pkt(sess, NULL);
session_set_cur_dir(sess, SESSION_DIR_NONE);
}; };
} }
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;
}

32
src/session/session_manager.h
View File

@@ -8,24 +8,40 @@ extern "C"
#include "session.h" #include "session.h"
/* // #define SESSION_MANAGER_LOG_ERROR(format, ...) void(0)
* session manager = session pool + session table + session timer #ifndef SESSION_MANAGER_LOG_ERROR
* #define SESSION_MANAGER_LOG_ERROR(format, ...) \
* session pool : alloc and free session fprintf(stderr, "ERROR (session manager), " format "\n", ##__VA_ARGS__);
* session table : find session by session key #endif
* session timer : session timeout // #define SESSION_MANAGER_LOG_DEBUG(format, ...) void(0)
* session manager: manage session pool, session table and session timer #ifndef SESSION_MANAGER_LOG_DEBUG
*/ #define SESSION_MANAGER_LOG_DEBUG(format, ...) \
fprintf(stderr, "DEBUG (session manager), " format "\n", ##__VA_ARGS__);
#endif
// create and destroy
struct session_manager; struct session_manager;
struct session_manager *session_manager_create(uint64_t max_session_num); struct session_manager *session_manager_create(uint64_t max_session_num);
void session_manager_destroy(struct session_manager *mgr); void session_manager_destroy(struct session_manager *mgr);
// config
typedef void (*session_event_cb)(struct session *sess, uint32_t event, void *arg); typedef void (*session_event_cb)(struct session *sess, uint32_t event, void *arg);
void session_manager_set_session_eventcb(struct session_manager *mgr, session_event_cb cb, void *arg); void session_manager_set_session_eventcb(struct session_manager *mgr, session_event_cb cb, void *arg);
void session_manager_set_packet_timeout(struct session_manager *mgr, uint64_t timeout_ms);
void session_manager_set_closing_timeout(struct session_manager *mgr, uint64_t timeout_ms);
// core functions
struct session *session_manager_find_session(struct session_manager *mgr, const struct packet *pkt); struct session *session_manager_find_session(struct session_manager *mgr, const struct packet *pkt);
void session_manager_dispatch(struct session_manager *mgr); void session_manager_dispatch(struct session_manager *mgr);
// for debug
uint64_t session_manager_get_tcp_opening_sess_num(struct session_manager *mgr);
uint64_t session_manager_get_tcp_closing_sess_num(struct session_manager *mgr);
uint64_t session_manager_get_tcp_active_sess_num(struct session_manager *mgr);
uint64_t session_manager_get_udp_opening_sess_num(struct session_manager *mgr);
uint64_t session_manager_get_udp_closing_sess_num(struct session_manager *mgr);
uint64_t session_manager_get_udp_active_sess_num(struct session_manager *mgr);
#ifdef __cpluscplus #ifdef __cpluscplus
} }
#endif #endif

31
src/session/session_private.h
View File

@@ -17,6 +17,29 @@ extern "C"
#define EX_DATA_MAX_COUNT 128 #define EX_DATA_MAX_COUNT 128
#define SESSION_EVENT_QUEUE_SIZE 256 #define SESSION_EVENT_QUEUE_SIZE 256
enum tcp_ex_data
{
// HANDSHAKE
TCP_SYN_RECVED = 1 << 0,
TCP_SYNACK_RECVED = 1 << 1,
TCP_ACK_RECVED = 1 << 2,
// ESTABLISHED
TCP_C2S_PAYLOAD_RECVED = 1 << 3,
TCP_S2C_PAYLOAD_RECVED = 1 << 4,
// FIN
TCP_C2S_FIN_RECVED = 1 << 5,
TCP_S2C_FIN_RECVED = 1 << 6,
// RST
TCP_C2S_RST_RECVED = 1 << 7,
TCP_S2C_RST_RECVED = 1 << 8,
};
enum udp_ex_data
{
UDP_C2S_RECVED = 1 << 0,
UDP_S2C_RECVED = 1 << 1,
};
struct event_queue struct event_queue
{ {
uint32_t head_idx; uint32_t head_idx;
@@ -92,6 +115,7 @@ struct session
// session table key // session table key
struct tuple6 tuple; struct tuple6 tuple;
enum session_dir tuple_dir;
struct session *next_ptr; struct session *next_ptr;
struct session *prev_ptr; struct session *prev_ptr;
@@ -100,12 +124,17 @@ struct session
UT_hash_handle hh; UT_hash_handle hh;
/****************************** /******************************
* Session Manager Zone * Session Queue Zone
******************************/ ******************************/
struct session *next_ready_ptr; struct session *next_ready_ptr;
}; };
// 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);
extern uint8_t tcp_builtin_ex;
extern uint8_t udp_builtin_ex;
#ifdef __cpluscplus #ifdef __cpluscplus
} }
#endif #endif

77
src/session/session_queue.cpp Normal file
View File

@@ -0,0 +1,77 @@
#include "session_queue.h"
#include "session_private.h"
struct session_queue
{
struct session *head;
struct session *tail;
int count;
};
struct session_queue *session_queue_create()
{
struct session_queue *queue = (struct session_queue *)calloc(1, sizeof(struct session_queue));
if (queue == NULL)
{
return NULL;
}
return queue;
}
void session_queue_destroy(struct session_queue *queue)
{
if (queue)
{
free(queue);
queue = NULL;
}
}
void session_queue_push(struct session_queue *queue, struct session *sess)
{
if (queue == NULL || sess == NULL)
{
return;
}
if (queue->head == NULL)
{
queue->head = sess;
queue->tail = sess;
}
else
{
queue->tail->next_ready_ptr = sess;
queue->tail = sess;
}
sess->next_ready_ptr = NULL;
queue->count++;
}
struct session *session_queue_pop(struct session_queue *queue)
{
if (queue == NULL)
{
return NULL;
}
struct session *sess = queue->head;
if (sess == NULL)
{
return NULL;
}
if (queue->head == queue->tail)
{
queue->head = NULL;
queue->tail = NULL;
}
else
{
queue->head = sess->next_ready_ptr;
}
sess->next_ready_ptr = NULL;
queue->count--;
return sess;
}

22
src/session/session_queue.h Normal file
View File

@@ -0,0 +1,22 @@
#ifndef _SESSION_QUEUE_H
#define _SESSION_QUEUE_H
#ifdef __cpluscplus
extern "C"
{
#endif
#include "session.h"
struct session_queue;
struct session_queue *session_queue_create();
void session_queue_destroy(struct session_queue *queue);
void session_queue_push(struct session_queue *queue, struct session *sess);
struct session *session_queue_pop(struct session_queue *queue);
#ifdef __cpluscplus
}
#endif
#endif

42
src/session/session_table.cpp
View File

@@ -12,8 +12,8 @@ struct session_table
void *arg; void *arg;
uint64_t count; uint64_t count;
struct session *oldest_ptr; struct session *least_recently_unused;
struct session *newest_ptr; struct session *least_recently_used;
}; };
/****************************************************************************** /******************************************************************************
@@ -52,19 +52,19 @@ static void session_table_add_session_to_linklist(struct session_table *table, s
return; return;
} }
if (table->newest_ptr == NULL) if (table->least_recently_used == NULL)
{ {
table->oldest_ptr = sess; table->least_recently_unused = sess;
table->newest_ptr = sess; table->least_recently_used = sess;
sess->prev_ptr = NULL; sess->prev_ptr = NULL;
sess->next_ptr = NULL; sess->next_ptr = NULL;
} }
else else
{ {
sess->next_ptr = table->newest_ptr; sess->next_ptr = table->least_recently_used;
table->newest_ptr->prev_ptr = sess; table->least_recently_used->prev_ptr = sess;
sess->prev_ptr = NULL; sess->prev_ptr = NULL;
table->newest_ptr = sess; table->least_recently_used = sess;
} }
} }
@@ -77,17 +77,17 @@ static void session_table_del_session_from_linklist(struct session_table *table,
if (sess->prev_ptr == NULL && sess->next_ptr == NULL) if (sess->prev_ptr == NULL && sess->next_ptr == NULL)
{ {
table->oldest_ptr = NULL; table->least_recently_unused = NULL;
table->newest_ptr = NULL; table->least_recently_used = NULL;
} }
else if (sess->prev_ptr == NULL && sess->next_ptr != NULL) else if (sess->prev_ptr == NULL && sess->next_ptr != NULL)
{ {
table->newest_ptr = sess->next_ptr; table->least_recently_used = sess->next_ptr;
sess->next_ptr->prev_ptr = NULL; sess->next_ptr->prev_ptr = NULL;
} }
else if (sess->prev_ptr != NULL && sess->next_ptr == NULL) else if (sess->prev_ptr != NULL && sess->next_ptr == NULL)
{ {
table->oldest_ptr = sess->prev_ptr; table->least_recently_unused = sess->prev_ptr;
sess->prev_ptr->next_ptr = NULL; sess->prev_ptr->next_ptr = NULL;
} }
else else
@@ -107,8 +107,8 @@ struct session_table *session_table_create()
{ {
struct session_table *table = (struct session_table *)calloc(1, sizeof(struct session_table)); struct session_table *table = (struct session_table *)calloc(1, sizeof(struct session_table));
table->count = 0; table->count = 0;
table->oldest_ptr = NULL; table->least_recently_unused = NULL;
table->newest_ptr = NULL; table->least_recently_used = NULL;
return table; return table;
} }
@@ -204,25 +204,31 @@ struct session *session_table_find_session(struct session_table *table, const st
struct session *sess = NULL; struct session *sess = NULL;
HASH_FIND(hh, table->root, tuple, sizeof(struct tuple6), sess); HASH_FIND(hh, table->root, tuple, sizeof(struct tuple6), sess);
if (sess)
{
session_table_del_session_from_linklist(table, sess);
session_table_add_session_to_linklist(table, sess);
}
return sess; return sess;
} }
struct session *session_table_find_oldest_session(struct session_table *table) struct session *session_table_find_least_recently_unused_session(struct session_table *table)
{ {
if (table == NULL) if (table == NULL)
{ {
return NULL; return NULL;
} }
return table->oldest_ptr; return table->least_recently_unused;
} }
struct session *session_table_find_newest_session(struct session_table *table) struct session *session_table_find_least_recently_used_session(struct session_table *table)
{ {
if (table == NULL) if (table == NULL)
{ {
return NULL; return NULL;
} }
return table->newest_ptr; return table->least_recently_used;
} }

4
src/session/session_table.h
View File

@@ -20,8 +20,8 @@ void session_table_set_freecb(struct session_table *table, session_free_cb free_
int session_table_add_session(struct session_table *table, const struct tuple6 *tuple, struct session *sess); int session_table_add_session(struct session_table *table, const struct tuple6 *tuple, struct session *sess);
void session_table_delete_session(struct session_table *table, const struct tuple6 *tuple); void session_table_delete_session(struct session_table *table, const struct tuple6 *tuple);
struct session *session_table_find_session(struct session_table *table, const struct tuple6 *tuple); struct session *session_table_find_session(struct session_table *table, const struct tuple6 *tuple);
struct session *session_table_find_oldest_session(struct session_table *table); struct session *session_table_find_least_recently_unused_session(struct session_table *table);
struct session *session_table_find_newest_session(struct session_table *table); struct session *session_table_find_least_recently_used_session(struct session_table *table);
#ifdef __cpluscplus #ifdef __cpluscplus
} }