✨ feat(plugin manager integration): packet and session exdata&mq

deps/bitmap/bitmap.c

@@ -46,8 +46,21 @@ void bitmap_free(struct bitmap *bmp) {
 	}
 }
 
-
+int bitmap_is_all_zero(struct bitmap *bmp, int x, int y, int length) {
-
+    if (x < 0 || y < 0 || x >= bmp->width || y >= bmp->height) {
+        return -1; // Return error code if coordinates are out of bounds
+    }
+    int idx = y * bmp->width + x;
+    if (idx + length > bmp->width * bmp->height) {
+        return -1; // Return error if range exceeds bitmap bounds
+    }
+    for (int i = 0; i < length; i++) {
+        if (bmp->data[(idx + i) / 8] & (1 << ((idx + i) % 8))) {
+            return 0; // Return 0 if any bit is not zero
+        }
+    }
+    return 1; // Return 1 if all bits are zero
+}
 
 int test_bitmap() {
     struct bitmap *bmp = bitmap_new(10, 5, 1); // Create a 10x5 bitmap

deps/bitmap/bitmap.h

@@ -1,5 +1,9 @@
+#pragma once
+
 struct bitmap;
 struct bitmap * bitmap_new(int width, int height, int value);
 int bitmap_set(struct bitmap *bmp, int x, int y, int value);
 int bitmap_get(struct bitmap *bmp, int x, int y);
-void bitmap_free(struct bitmap *bmp);
+void bitmap_free(struct bitmap *bmp);
+
+int bitmap_is_all_zero(struct bitmap *bmp, int x, int y, int length);

include/CMakeLists.txt

@@ -4,5 +4,5 @@ install(FILES stellar/tunnel.h DESTINATION include/stellar/ COMPONENT LIBRARIES)
 install(FILES stellar/packet.h DESTINATION include/stellar/ COMPONENT LIBRARIES)
 install(FILES stellar/session.h DESTINATION include/stellar/ COMPONENT LIBRARIES)
 install(FILES stellar/stellar.h DESTINATION include/stellar/ COMPONENT LIBRARIES)
-install(FILES stellar/session_mq.h DESTINATION include/stellar/ COMPONENT LIBRARIES)
+install(FILES stellar/stellar_mq.h DESTINATION include/stellar/ COMPONENT LIBRARIES)
-install(FILES stellar/session_exdata.h DESTINATION include/stellar/ COMPONENT LIBRARIES)
+install(FILES stellar/stellar_exdata.h DESTINATION include/stellar/ COMPONENT LIBRARIES)

include/stellar/session_exdata.h

@@ -1,17 +0,0 @@
-#pragma once
-
-#ifdef __cplusplus
-extern "C"
-{
-#endif
-
-#include "stellar.h"
-
-typedef void session_exdata_free(struct session *sess, int idx, void *ex_ptr, void *arg);
-int stellar_session_exdata_new_index(struct stellar *st, const char *name, session_exdata_free *free_func,void *arg); 
-int session_exdata_set(struct session *sess, int idx, void *ex_ptr);
-void *session_exdata_get(struct session *sess, int idx);
-
-#ifdef __cplusplus
-}
-#endif

include/stellar/session_mq.h

@@ -1,33 +0,0 @@
-#pragma once
-
-#ifdef __cplusplus
-extern "C"
-{
-#endif
-
-#include "stellar.h"
-
-//session mq
-typedef void msg_free_cb_func(void *msg, void *msg_free_arg);
-typedef void on_msg_cb_func(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env);
-
-//return topic_id
-int stellar_session_mq_create_topic(struct stellar *st, const char *topic_name, msg_free_cb_func *msg_free_cb, void *msg_free_arg);
-
-int stellar_session_mq_get_topic_id(struct stellar *st, const char *topic_name);
-
-int stellar_session_mq_update_topic(struct stellar *st, int topic_id, msg_free_cb_func *msg_free_cb, void *msg_free_arg);
-
-int stellar_session_mq_destroy_topic(struct stellar *st, int topic_id);
-
-//return 0 if success, otherwise return -1.
-int stellar_session_mq_subscribe(struct stellar *st, int topic_id, on_msg_cb_func *plugin_on_msg_cb, int plugin_id);
-
-int session_mq_publish_message(struct session *sess, int topic_id, void *msg);
-
-int session_mq_ignore_message(struct session *sess, int topic_id, int plugin_id);
-int session_mq_unignore_message(struct session *sess, int topic_id, int plugin_id);
-
-#ifdef __cplusplus
-}
-#endif

include/stellar/stellar.h

@@ -6,7 +6,6 @@ extern "C"
 #endif
 
 #include <stdint.h>
-#include "stellar/session.h"
 
 struct session;
 struct stellar;
@@ -48,6 +47,9 @@ typedef int plugin_on_polling_func(void *plugin_env);
 //return polling plugin_id
 int stellar_polling_plugin_register(struct stellar *st,  plugin_on_polling_func on_polling, void *plugin_env);
 
+void stellar_emit_datapath_telemetry(struct packet *pkt,  const char * module, const char *str);
+
+int stellar_get_worker_thread_num(struct stellar *st);
 uint16_t stellar_get_current_thread_index();
 // only send user crafted packet, can't send packet which come from network
 void stellar_send_crafted_packet(struct stellar *st, struct packet *pkt);

include/stellar/stellar_exdata.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include "stellar.h"
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+typedef void stellar_exdata_free(int idx, void *ex_ptr, void *arg);
+int stellar_exdata_new_index(struct stellar *st, const char *name, stellar_exdata_free *free_func,void *arg); 
+
+//packet exdata api
+int packet_exdata_set(struct packet *pkt, int idx, void *ex_ptr);
+void *packet_exdata_get(struct packet *pkt, int idx);
+
+//session exdata api
+int session_exdata_set(struct session *sess, int idx, void *ex_ptr);
+void *session_exdata_get(struct session *sess, int idx);
+
+#ifdef __cplusplus
+}
+#endif

include/stellar/stellar_mq.h

@@ -0,0 +1,51 @@
+#pragma once
+#include "stellar.h"
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+//topic api
+typedef void stellar_msg_free_cb_func(void *msg, void *msg_free_arg);
+
+//return topic_id
+int stellar_mq_create_topic(struct stellar *st, const char *topic_name, stellar_msg_free_cb_func *msg_free_cb, void *msg_free_arg);
+int stellar_mq_get_topic_id(struct stellar *st, const char *topic_name);
+int stellar_mq_update_topic(struct stellar *st, int topic_id, stellar_msg_free_cb_func *msg_free_cb, void *msg_free_arg);
+int stellar_mq_destroy_topic(struct stellar *st, int topic_id);
+
+
+enum stellar_mq_priority 
+{
+	STELLAR_MQ_PRIORITY_LOW,
+	STELLAR_MQ_PRIORITY_NORMAL,
+	STELLAR_MQ_PRIORITY_HIGH,
+	STELLAR_MQ_PRIORITY_MAX,
+};
+
+//session mq api
+typedef void on_session_msg_cb_func(struct session *sess, int topic_id, const void *msg, void *per_session_ctx, void *plugin_env);
+
+//return 0 if success, otherwise return -1.
+int stellar_session_mq_subscribe(struct stellar *st, int topic_id, on_session_msg_cb_func *plugin_on_msg_cb, int plugin_id);
+int session_mq_publish_message(struct session *sess, int topic_id, void *msg);
+int session_mq_publish_message_with_priority(struct session *sess, int topic_id, void *msg, enum stellar_mq_priority priority);
+
+int session_mq_ignore_message(struct session *sess, int topic_id, int plugin_id);
+int session_mq_unignore_message(struct session *sess, int topic_id, int plugin_id);
+
+int session_mq_topic_is_active(struct session *sess, int topic_id);
+
+
+//packet mq api
+
+typedef void on_packet_msg_cb_func(struct packet *pkt, int topic_id, const void *msg, void *plugin_env);
+//return 0 if success, otherwise return -1.
+int stellar_packet_mq_subscribe(struct stellar *st, int topic_id, on_packet_msg_cb_func *plugin_on_msg_cb, int plugin_id); //packet plugin only
+int packet_mq_publish_message(struct packet *pkt, int topic_id, void *msg);
+int packet_mq_publish_message_with_priority(struct packet *pkt, int topic_id, void *msg, enum stellar_mq_priority priority);
+
+#ifdef __cplusplus
+}
+#endif

include/stellar/utils.h

@@ -31,6 +31,14 @@
 	(type *)( (char *)__mptr - offsetof(type,member) );})
 #endif
 
+#ifndef likely
+#define likely(x)  __builtin_expect((x),1)
+#endif /* likely */
+
+#ifndef unlikely
+#define unlikely(x)  __builtin_expect((x),0)
+#endif /* unlikely */
+
 #ifndef __unused
 #define __unused __attribute__((__unused__))
 #endif

src/id_generator/CMakeLists.txt

@@ -1,4 +1,5 @@
 add_library(id_generator id_generator.cpp)
 target_include_directories(id_generator PUBLIC ${CMAKE_CURRENT_LIST_DIR})
+target_include_directories(id_generator PUBLIC ${CMAKE_SOURCE_DIR}/include/stellar)
 target_include_directories(id_generator PUBLIC ${CMAKE_SOURCE_DIR}/src/stellar)
-target_link_libraries(id_generator log stellar_core)
+target_link_libraries(id_generator PRIVATE stellar_core log)

src/id_generator/id_generator.cpp

@@ -3,7 +3,7 @@
 
 #include "log.h"
 #include "macro.h"
-#include "stellar_core.h"
+#include "stellar.h"
 #include "id_generator.h"
 
 #define ID_GENERATOR_LOG_ERROR(format, ...) LOG_ERROR("id generator", format, ##__VA_ARGS__)
@@ -66,7 +66,7 @@ uint64_t id_generator_alloc(uint64_t now_sec)
 #define MAX_ID_PER_THREAD (32768)
 #define MAX_ID_BASE_TIME (268435456L)
 
-    uint64_t thr_idx = stellar_get_current_thread_index();
+    uint64_t thr_idx = (uint16_t)stellar_get_current_thread_index();
 
     uint64_t global_id = 0;
     uint64_t id_per_thread = (global_id_generator.thread_volatile[thr_idx]++) % MAX_ID_PER_THREAD;

src/packet/packet_def.h

@@ -44,6 +44,7 @@ struct raw_layer
 
 struct packet
 {
+    void * user_data;
     struct raw_layer layers[PACKET_MAX_LAYERS];
     struct raw_layer *frag_layer; // fragment layer
     int8_t layers_used;

src/packet/packet_utils.cpp

@@ -546,3 +546,13 @@ void layer_convert(const struct raw_layer *in, struct layer *out)
     out->hdr_len = in->hdr_len;
     out->hdr.raw = (char *)in->hdr_ptr;
 }
+
+void packet_set_user_data(struct packet *pkt, void *data)
+{
+    pkt->user_data=data;
+}
+
+void *packet_get_user_data(struct packet *pkt)
+{
+    return pkt->user_data;
+}

src/packet/packet_utils.h

@@ -36,6 +36,9 @@ void packet_set_action(struct packet *pkt, enum packet_action action);
 enum packet_action packet_get_action(const struct packet *pkt);
 
 
+void *packet_get_user_data(struct packet *pkt);
+void packet_set_user_data(struct packet *pkt, void *data);
+
 /******************************************************************************
  * tuple uitls
  ******************************************************************************/

src/plugin/CMakeLists.txt

@@ -1,4 +1,8 @@
 add_library(plugin_manager plugin_manager.cpp)
 target_include_directories(plugin_manager PUBLIC ${CMAKE_CURRENT_LIST_DIR})
+target_include_directories(plugin_manager PUBLIC ${CMAKE_SOURCE_DIR}/include/)
+target_include_directories(plugin_manager PUBLIC ${CMAKE_SOURCE_DIR}/src/)
 target_include_directories(plugin_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/)
-target_link_libraries(plugin_manager bitmap toml session_manager stellar_core ${CMAKE_DL_LIBS})
+target_link_libraries(plugin_manager bitmap toml ${CMAKE_DL_LIBS})
+
+add_subdirectory(test)

src/plugin/plugin_manager.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include "stellar/stellar.h"
+
 #ifdef __cplusplus
 extern "C"
 {
@@ -11,18 +13,19 @@ struct plugin_manager_runtime;
 struct plugin_manager_schema *plugin_manager_init(struct stellar *st, const char *plugin_spec_file_path);
 void plugin_manager_exit(struct plugin_manager_schema *plug_mgr);
 
-void plugin_manager_on_packet(struct plugin_manager_schema *plug_mgr, struct packet *pkt);
+void plugin_manager_on_packet_ingress(struct plugin_manager_schema *plug_mgr, struct packet *pkt);
-
+void plugin_manager_on_packet_egress(struct plugin_manager_schema *plug_mgr, struct packet *pkt);
 //return polling work state, 0: idle, 1: working
 int plugin_manager_on_polling(struct plugin_manager_schema *plug_mgr);
 
 //publish and dispatch session msg(msg, pkt) on session_mq
-void plugin_manager_on_session_ingress(struct session *sess,const struct packet *pkt);
+void plugin_manager_on_session_ingress(struct session *sess,struct packet *pkt);
-void plugin_manager_on_session_egress(struct session *sess,const struct packet *pkt);
+void plugin_manager_on_session_egress(struct session *sess,struct packet *pkt);
+void plugin_manager_on_session_closing(struct session *sess);
 
 struct plugin_manager_runtime *plugin_manager_session_runtime_new(struct plugin_manager_schema *plug_mgr, struct session *sess);
 void plugin_manager_session_runtime_free(struct plugin_manager_runtime *plug_mgr_rt);
 
 #ifdef __cplusplus
 }
-#endif
+#endif

src/plugin/plugin_manager_interna.h

@@ -0,0 +1,192 @@
+#pragma once
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include "plugin_manager.h"
+
+#include "stellar/stellar.h"
+#include "stellar/stellar_mq.h"
+#include "stellar/stellar_exdata.h"
+
+#include "bitmap/bitmap.h"
+#include "uthash/utarray.h"
+
+struct per_thread_exdata_array
+{
+	struct stellar_exdata *exdata_array;
+};
+
+struct stellar_message;
+
+struct plugin_manger_per_thread_data
+{
+	struct per_thread_exdata_array per_thread_pkt_exdata_array;
+	struct stellar_message *priority_mq[STELLAR_MQ_PRIORITY_MAX];// message list
+	struct stellar_message *dealth_letter_queue;// dlq list
+	long long pub_packet_msg_cnt;
+};
+
+
+
+struct plugin_manager_schema
+{
+	struct stellar *st;
+	UT_array *plugin_load_specs_array;
+	UT_array *stellar_exdata_schema_array;
+	UT_array *stellar_mq_schema_array;
+	UT_array *registered_session_plugin_array;
+	UT_array *registered_packet_plugin_array;
+	UT_array *registered_polling_plugin_array;
+	int stellar_mq_topic_num;
+	int packet_topic_subscriber_num;
+	int session_topic_subscriber_num;
+	int tcp_topic_id;
+	int tcp_stream_topic_id;
+	int udp_topic_id;
+	int egress_topic_id;
+	int control_packet_topic_id;
+	int max_message_dispatch;
+	struct plugin_manger_per_thread_data *per_thread_data;
+}__attribute__((aligned(sizeof(void*))));
+
+enum plugin_exdata_state
+{ INIT, ACTIVE, EXIT };
+
+struct stellar_exdata
+{
+	void *exdata;
+	enum plugin_exdata_state state;
+};
+
+
+
+struct stellar_exdata_schema
+{
+	char *name;
+	stellar_exdata_free *free_func;
+	
+	void *free_arg;
+	int idx;
+}__attribute__((aligned(sizeof(void*))));
+
+
+enum stellar_topic_type
+{
+	ON_SESSION_TOPIC,
+	ON_PACKET_TOPIC,
+};
+
+struct stellar_message
+{
+    struct
+    {
+        int topic_id;
+		enum stellar_topic_type type;
+		enum stellar_mq_priority priority;
+    } header;
+    void *body;
+    struct stellar_message *next, *prev;
+} __attribute__((aligned(sizeof(void *))));
+
+typedef struct stellar_mq_subscriber 
+{
+	int topic_subscriber_idx;
+	int plugin_idx;
+	union 
+	{
+    	on_session_msg_cb_func *sess_msg_cb;
+		on_packet_msg_cb_func *pkt_msg_cb;
+		void *msg_cb;
+	};
+    struct stellar_mq_subscriber *next, *prev;
+}stellar_mq_subscriber __attribute__((aligned(sizeof(void*))));
+
+
+struct stellar_mq_topic_schema
+{
+	char *topic_name;
+	void *free_cb_arg;
+	int topic_id;
+	int subscriber_cnt;
+	int is_destroyed;
+	stellar_msg_free_cb_func *free_cb;
+	struct stellar_mq_subscriber *subscribers; 
+}__attribute__((aligned(sizeof(void*))));
+
+
+
+struct session_plugin_ctx_runtime
+{
+	enum plugin_exdata_state state;
+	int session_plugin_id;
+	void *plugin_ctx;
+}__attribute__((aligned(sizeof(void*))));
+
+
+
+struct plugin_manager_runtime
+{
+	struct plugin_manager_schema *plug_mgr;
+	struct session *sess;
+	struct bitmap *session_mq_status; //N * M bits, N topic, M subscriber
+	struct bitmap *session_topic_status; //N bits, N topic
+	struct stellar_exdata *sess_exdata_array;
+	struct session_plugin_ctx_runtime *plugin_ctx_array;//N plugins TODO: call alloc and free
+	int current_session_plugin_id;
+	int pub_session_msg_cnt;
+}__attribute__((aligned(sizeof(void*))));
+
+struct registered_packet_plugin_schema
+{
+	char ip_protocol;
+	plugin_on_packet_func *on_packet;
+	void *plugin_env;
+	UT_array *registed_packet_mq_subscriber_info;
+}__attribute__((aligned(sizeof(void*))));
+
+struct registered_polling_plugin_schema
+{
+	plugin_on_polling_func *on_polling;
+	void *plugin_env;
+}__attribute__((aligned(sizeof(void*))));
+
+struct stellar_mq_subscriber_info
+{
+	int topic_id;
+	int subscriber_idx;
+}__attribute__((aligned(sizeof(void*))));
+
+struct registered_session_plugin_schema
+{
+	session_ctx_new_func *on_ctx_new;
+	session_ctx_free_func *on_ctx_free;
+	void *plugin_env;
+	UT_array *registed_session_mq_subscriber_info;
+}__attribute__((aligned(sizeof(void*))));
+
+#define SESSION_PULGIN_ID_BASE 0x00000
+#define PACKET_PULGIN_ID_BASE 0x10000
+#define POLLING_PULGIN_ID_BASE 0x20000
+
+/*******************************
+ * PLUGIN MANAGER INIT & EXIT *
+ *******************************/
+
+#define MAX_MSG_PER_DISPATCH 128
+
+#include <dlfcn.h>
+
+struct plugin_specific
+{
+	char plugin_name[256];
+	plugin_on_load_func *load_cb;
+	plugin_on_unload_func *unload_cb;
+	void *plugin_ctx;
+}__attribute__((aligned(sizeof(void*))));
+
+#ifdef __cplusplus
+}
+#endif

src/plugin/test/CMakeLists.txt

@@ -0,0 +1,17 @@
+add_executable(gtest_plugin_manager
+plugin_manager_gtest_main.cpp
+)
+
+include_directories(${CMAKE_SOURCE_DIR}/src/plugin/)
+
+target_link_libraries(
+  gtest_plugin_manager
+  plugin_manager
+  dl
+  "-rdynamic"
+  gtest
+  gmock
+)
+
+include(GoogleTest)
+gtest_discover_tests(gtest_plugin_manager)

src/plugin/test/plugin_manager_gtest_mock.h

@@ -0,0 +1,116 @@
+#pragma once
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include "../plugin_manager_interna.h"
+
+#include "stellar/session.h"
+#include "tuple/tuple.h"
+
+//mock stellar
+struct stellar
+{
+	struct plugin_manager_schema *plug_mgr;
+};
+
+enum packet_type
+{
+	UNKNOWN,
+	IPv4,
+	IPv6,
+	UDP,
+	TCP,
+	TCP_STREAM,
+	CONTROL,
+};
+
+struct packet
+{
+	struct stellar *st;
+	enum packet_type type;
+	unsigned char ip_proto;
+};
+
+
+struct session
+{
+	struct plugin_manager_runtime *plug_mgr_rt;
+	enum session_type type;
+	enum session_state state;
+	int sess_pkt_cnt;
+};
+
+struct plugin_manager_schema * stellar_get_plugin_manager(struct stellar *st)
+{
+	return st->plug_mgr;
+}
+
+int stellar_set_plugin_manger(struct stellar *st, struct plugin_manager_schema *pm)
+{
+	st->plug_mgr=pm;
+	return 0;
+}
+
+int stellar_get_worker_thread_num(struct stellar *st)
+{
+	return 16;
+}
+
+uint16_t stellar_get_current_thread_index() 
+{
+	return 0;
+}
+
+unsigned char packet_get_ip_protocol(struct packet *pkt)
+{
+	return pkt->ip_proto;
+}
+
+enum session_type session_get_type(const struct session *sess)
+{
+	return sess->type;
+
+}
+
+void session_set_user_data(struct session *sess, void *user_data)
+{
+    sess->plug_mgr_rt = (struct plugin_manager_runtime *)user_data;
+}
+
+void *session_get_user_data(const struct session *sess)
+{
+    return sess->plug_mgr_rt;
+}
+
+void *packet_get_user_data(const struct packet *pkt)
+{
+	return pkt->st->plug_mgr;
+}
+
+int packet_get_innermost_tuple6(const struct packet *pkt, struct tuple6 *tuple)
+{
+	tuple->ip_proto = pkt->ip_proto;
+	return 0;
+}
+
+uint8_t packet_is_ctrl(const struct packet *pkt)
+{
+	return 0;
+}
+
+struct tcp_segment *session_get_tcp_segment(struct session *sess)
+{
+	return NULL;
+}
+
+void session_free_tcp_segment(struct session *sess, struct tcp_segment *seg)
+{
+	return;
+}
+
+#ifdef __cplusplus
+}
+#endif

src/stellar/CMakeLists.txt

@@ -1,16 +1,17 @@
 set(SOURCE stellar_config.cpp stellar_stat.cpp stellar_core.cpp)
-set(LIBRARY times plugin_manager session_manager ip_reassembly packet_io pthread fieldstat4 toml)
+set(LIBRARY times session_manager plugin_manager ip_reassembly packet_io packet pthread fieldstat4 toml)
 
 add_library(stellar_core STATIC ${SOURCE})
-target_link_libraries(stellar_core ${LIBRARY})
+target_link_libraries(stellar_core PRIVATE ${LIBRARY})
 
 add_library(stellar_devel SHARED ${SOURCE})
-target_link_libraries(stellar_devel ${LIBRARY})
+#target_link_libraries(stellar_devel ${LIBRARY})
 set_target_properties(stellar_devel PROPERTIES LINK_FLAGS "-Wl,--version-script=${CMAKE_CURRENT_LIST_DIR}/version.map")
+target_link_libraries(stellar_devel PRIVATE -Wl,--whole-archive ${LIBRARY} -Wl,--no-whole-archive)
 
 add_executable(stellar main.cpp)
-target_link_libraries(stellar stellar_core)
+target_link_libraries(stellar PRIVATE -Wl,--whole-archive stellar_core ${LIBRARY} -Wl,--no-whole-archive)
-target_link_libraries(stellar "-rdynamic")
+target_link_libraries(stellar PRIVATE "-rdynamic")
 set_target_properties(stellar PROPERTIES LINK_FLAGS "-Wl,--version-script=${CMAKE_CURRENT_LIST_DIR}/version.map")
 
 install(TARGETS stellar RUNTIME DESTINATION bin COMPONENT PROGRAM)

src/stellar/stellar_core.cpp

@@ -101,6 +101,7 @@ static inline void free_evicted_sessions(struct session_manager *sess_mgr, uint6
         if (sess)
         {
             plugin_ctx = session_get_user_data(sess);
+            plugin_manager_on_session_closing(sess);
             plugin_manager_session_runtime_free((struct plugin_manager_runtime *)plugin_ctx);
             session_manager_free_session(sess_mgr, sess);
         }
@@ -121,6 +122,7 @@ static inline void free_expired_sessions(struct session_manager *sess_mgr, uint6
         if (sess)
         {
             plugin_ctx = session_get_user_data(sess);
+            plugin_manager_on_session_closing(sess);
             plugin_manager_session_runtime_free((struct plugin_manager_runtime *)plugin_ctx);
             session_manager_free_session(sess_mgr, sess);
         }
@@ -164,6 +166,11 @@ static void *work_thread(void *arg)
 
     memset(packets, 0, sizeof(packets));
 
+    for(int i=0; i<RX_BURST_MAX; i++)
+    {
+       packet_set_user_data(&packets[i], (void *)plug_mgr);
+    }
+
     snprintf(thd_name, sizeof(thd_name), "stellar:%d", thr_idx);
     prctl(PR_SET_NAME, (unsigned long long)thd_name, NULL, NULL, NULL);
 
@@ -200,7 +207,7 @@ static void *work_thread(void *arg)
             defraged_pkt = NULL;
             pkt = &packets[i];
 
-            plugin_manager_on_packet(plug_mgr, pkt);
+            plugin_manager_on_packet_ingress(plug_mgr, pkt);
             if (packet_is_fragment(pkt))
             {
                 defraged_pkt = ip_reassembly_packet(ip_reass, pkt, now_ms);
@@ -211,10 +218,12 @@ static void *work_thread(void *arg)
                 else
                 {
                     pkt = defraged_pkt;
-                    plugin_manager_on_packet(plug_mgr, pkt);
+                    plugin_manager_on_packet_ingress(plug_mgr, pkt);
+                    plugin_manager_on_packet_egress(plug_mgr, pkt);
                 }
             }
 
+            pkt = &packets[i];
             sess = session_manager_lookup_session_by_packet(sess_mgr, pkt);
             if (sess == NULL)
             {
@@ -241,6 +250,8 @@ static void *work_thread(void *arg)
 
         fast_path:
             plugin_manager_on_session_egress(sess, pkt);
+            plugin_manager_on_packet_egress(plug_mgr, pkt);
+
             if (sess && session_get_current_state(sess) == SESSION_STATE_DISCARD)
             {
                 packet_set_action(pkt, PACKET_ACTION_DROP);
@@ -567,4 +578,9 @@ void stellar_send_crafted_packet(struct stellar *st, struct packet *pkt)
     {
         packet_io_inject(packet_io, thr_idx, pkt, 1);
     }
+}
+
+int stellar_get_worker_thread_num(struct stellar *st)
+{
+    return st->config.pkt_io_opts.nr_threads;
 }

src/stellar/stellar_core.h

@@ -15,7 +15,6 @@ void stellar_set_plugin_manger(struct stellar *st, struct plugin_manager_schema
 
 // only send user crafted packet, can't send packet which come from network
 void stellar_send_crafted_packet(struct stellar *st, struct packet *pkt);
-int stellar_run(int argc, char **argv);
 
 #ifdef __cplusplus
 }

src/stellar/version.map

@@ -26,10 +26,10 @@ global:
 	session_exdata_set;
 	session_exdata_get;
 
-	stellar_session_mq_create_topic;
+	stellar_mq_create_topic;
-	stellar_session_mq_get_topic_id;
+	stellar_mq_get_topic_id;
-	stellar_session_mq_update_topic;
+	stellar_mq_update_topic;
-	stellar_session_mq_destroy_topic;
+	stellar_mq_destroy_topic;
 	stellar_session_mq_subscribe;
 	session_mq_publish_message;
 	session_mq_ignore_message;

test/packet_inject/CMakeLists.txt

@@ -1,14 +1,15 @@
 # build libpacket_inject.so
 add_library(packet_inject SHARED packet_inject.cpp)
-target_link_libraries(packet_inject stellar_devel toml)
+target_link_libraries(packet_inject toml)
 target_include_directories(packet_inject PUBLIC ${CMAKE_SOURCE_DIR}/include/)
 set_target_properties(packet_inject PROPERTIES LINK_FLAGS "-Wl,--version-script=${CMAKE_CURRENT_LIST_DIR}/version.map")
 
 # build gtest
 function(packet_inject_add_case EXEC_NAME)
     add_executable(${EXEC_NAME} ${EXEC_NAME}.cpp)
-    target_link_libraries(${EXEC_NAME} "-rdynamic")
+    target_include_directories(${EXEC_NAME} PUBLIC ${CMAKE_SOURCE_DIR}/include/)
-    target_link_libraries(${EXEC_NAME} stellar_devel gtest)
+    target_link_libraries(${EXEC_NAME} PRIVATE "-rdynamic")
+    target_link_libraries(${EXEC_NAME} PRIVATE stellar_devel gtest)
     gtest_discover_tests(${EXEC_NAME})
 endfunction()
 

test/packet_inject/packet_inject.cpp

@@ -5,8 +5,10 @@
 #include <arpa/inet.h>
 
 #include "toml.h"
+#include "stellar/stellar.h"
 #include "stellar/layer.h"
-#include "stellar/session_mq.h"
+#include "stellar/session.h"
+#include "stellar/stellar_mq.h"
 
 #define LOG_ERR(fmt, ...) printf("ERROR [packet inject] " fmt, ##__VA_ARGS__)
 #define LOG_INFO(fmt, ...) printf("INFO [packet inject] " fmt, ##__VA_ARGS__)
@@ -545,8 +547,8 @@ extern "C"
 
         ctx->st = st;
         ctx->sess_plug_id = stellar_session_plugin_register(st, on_sess_new, on_sess_free, ctx);
-        ctx->tcp_topic_id = stellar_session_mq_get_topic_id(st, TOPIC_TCP);
+        ctx->tcp_topic_id = stellar_mq_get_topic_id(st, TOPIC_TCP);
-        ctx->udp_topic_id = stellar_session_mq_get_topic_id(st, TOPIC_UDP);
+        ctx->udp_topic_id = stellar_mq_get_topic_id(st, TOPIC_UDP);
 
         stellar_session_mq_subscribe(st, ctx->tcp_topic_id, on_sess_msg, ctx->sess_plug_id);
         stellar_session_mq_subscribe(st, ctx->udp_topic_id, on_sess_msg, ctx->sess_plug_id);