test(packet manager): test packets flow between multiple queues

5 files changed, 322 insertions, 126 deletions

diff --git a/infra/packet_manager/CMakeLists.txt b/infra/packet_manager/CMakeLists.txt
index f764a51..ada4303 100644
--- a/infra/packet_manager/CMakeLists.txt
+++ b/infra/packet_manager/CMakeLists.txt
@@ -11,6 +11,6 @@ target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/uthash
 target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/logger)
 target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/include)
 target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/infra)
-target_link_libraries(packet_manager tuple logger dablooms)
+target_link_libraries(packet_manager tuple logger dablooms mq)
 
 add_subdirectory(test)
\ No newline at end of file
diff --git a/infra/packet_manager/packet_manager.c b/infra/packet_manager/packet_manager.c
index efc14a7..b13802f 100644
--- a/infra/packet_manager/packet_manager.c
+++ b/infra/packet_manager/packet_manager.c
@@ -1,9 +1,12 @@
+#include <assert.h>
+
 #include "utils.h"
 #include "stellar/mq.h"
 #include "packet_private.h"
 #include "packet_manager_private.h"
 
 #define PACKET_MANAGER_LOG_ERROR(format, ...) STELLAR_LOG_ERROR(__thread_local_logger, "packet manager", format, ##__VA_ARGS__)
+#define PACKET_MANAGER_LOG_DEBUG(format, ...) STELLAR_LOG_DEBUG(__thread_local_logger, "packet manager", format, ##__VA_ARGS__)
 #define PACKET_MANAGER_LOG_INFO(format, ...) STELLAR_LOG_WARN(__thread_local_logger, "packet manager", format, ##__VA_ARGS__)
 
 TAILQ_HEAD(packet_queue, packet);
@@ -32,6 +35,7 @@ struct packet_manager_schema
 
 struct packet_manager_runtime
 {
+    uint16_t idx;
     struct mq_runtime *mq;
     struct packet_queue queue[PACKET_STAGE_MAX];
     struct packet_manager_stat stat;
@@ -45,6 +49,30 @@ struct packet_manager
 };
 
 /******************************************************************************
+ * packet stage
+ ******************************************************************************/
+
+const char *packet_stage_to_str(enum packet_stage stage)
+{
+    switch (stage)
+    {
+    case PACKET_STAGE_PREROUTING:
+        return "prerouting";
+    case PACKET_STAGE_INPUT:
+        return "input";
+    case PACKET_STAGE_FORWARD:
+        return "forward";
+    case PACKET_STAGE_OUTPUT:
+        return "output";
+    case PACKET_STAGE_POSTROUTING:
+        return "postrouting";
+    default:
+        assert(0);
+        return "unknown";
+    }
+}
+
+/******************************************************************************
  * packet manager config
  ******************************************************************************/
 
@@ -66,12 +94,14 @@ static struct packet_manager_config *packet_manager_config_new(const char *toml_
         return NULL;
     }
 
-    if (load_and_validate_toml_integer_config(toml_file, "packet_io.nr_worker_thread", (uint64_t *)&cfg->nr_worker_thread, 1, MAX_THREAD_NUM) != 0)
+    uint64_t val = 0;
+    if (load_and_validate_toml_integer_config(toml_file, "packet_io.nr_worker_thread", &val, 1, MAX_THREAD_NUM) != 0)
     {
         PACKET_MANAGER_LOG_ERROR("failed to load packet_io.nr_worker_thread from %s", toml_file);
         free(cfg);
         return NULL;
     }
+    cfg->nr_worker_thread = val;
 
     return cfg;
 }
@@ -80,8 +110,11 @@ static struct packet_manager_config *packet_manager_config_new(const char *toml_
  * packet manager schema
  ******************************************************************************/
 
-static void on_packet_stage_dispatch(int topic_id, void *msg, on_msg_cb_func *cb, void *cb_arg, void *dispatch_arg)
+static void on_packet_stage_dispatch(int topic_id, const void *msg, on_msg_cb_func *cb, void *cb_arg, void *dispatch_arg)
 {
+    assert(msg);
+    assert(dispatch_arg);
+
     struct packet_manager_schema *schema = (struct packet_manager_schema *)dispatch_arg;
     struct packet *pkt = (struct packet *)msg;
 
@@ -98,69 +131,59 @@ static void on_packet_stage_dispatch(int topic_id, void *msg, on_msg_cb_func *cb
     ((on_packet_stage_callback *)cb)(stage, pkt, cb_arg);
 }
 
-static void packet_manager_schema_free(struct packet_manager_schema *schema)
+static void packet_manager_schema_free(struct packet_manager_schema *pkt_mgr_schema)
 {
-    if (schema)
+    if (pkt_mgr_schema)
     {
-        if (schema->mq)
+        if (pkt_mgr_schema->mq)
         {
             for (int i = 0; i < PACKET_STAGE_MAX; i++)
             {
-                if (schema->topic_id[i] != -1)
+                if (pkt_mgr_schema->topic_id[i] >= 0)
                 {
-                    mq_schema_destroy_topic(schema->mq, schema->topic_id[i]);
+                    mq_schema_destroy_topic(pkt_mgr_schema->mq, pkt_mgr_schema->topic_id[i]);
                 }
             }
-
-            mq_schema_free(schema->mq);
         }
 
-        free(schema);
-        schema = NULL;
+        free(pkt_mgr_schema);
+        pkt_mgr_schema = NULL;
     }
 }
 
-static struct packet_manager_schema *packet_manager_schema_new()
+static struct packet_manager_schema *packet_manager_schema_new(struct mq_schema *mq)
 {
-    struct packet_manager_schema *schema = calloc(1, sizeof(struct packet_manager_schema));
-    if (schema == NULL)
+    struct packet_manager_schema *pkt_mgr_schema = calloc(1, sizeof(struct packet_manager_schema));
+    if (pkt_mgr_schema == NULL)
     {
         PACKET_MANAGER_LOG_ERROR("failed to allocate memory for packet_manager_schema");
         return NULL;
     }
 
-    schema->mq = mq_schema_new();
-    if (schema->mq == NULL)
-    {
-        PACKET_MANAGER_LOG_ERROR("failed to create mq_schema");
-        goto error_out;
-    }
+    pkt_mgr_schema->mq = mq;
 
     for (int i = 0; i < PACKET_STAGE_MAX; i++)
     {
-        schema->topic_id[i] = -1;
-    }
-    schema->topic_id[PACKET_STAGE_PREROUTING] = mq_schema_create_topic(schema->mq, "packet_stage_prerouting", on_packet_stage_dispatch, schema, NULL, NULL);
-    schema->topic_id[PACKET_STAGE_INPUT] = mq_schema_create_topic(schema->mq, "packet_stage_input", on_packet_stage_dispatch, schema, NULL, NULL);
-    schema->topic_id[PACKET_STAGE_FORWARD] = mq_schema_create_topic(schema->mq, "packet_stage_forward", on_packet_stage_dispatch, schema, NULL, NULL);
-    schema->topic_id[PACKET_STAGE_OUTPUT] = mq_schema_create_topic(schema->mq, "packet_stage_output", on_packet_stage_dispatch, schema, NULL, NULL);
-    schema->topic_id[PACKET_STAGE_POSTROUTING] = mq_schema_create_topic(schema->mq, "packet_stage_postrouting", on_packet_stage_dispatch, schema, NULL, NULL);
-    for (int i = 0; i < PACKET_STAGE_MAX; i++)
-    {
-        if (schema->topic_id[i] < 0)
+        pkt_mgr_schema->topic_id[i] = mq_schema_create_topic(pkt_mgr_schema->mq, packet_stage_to_str(i), (on_msg_dispatch_cb_func *)on_packet_stage_dispatch, pkt_mgr_schema, NULL, NULL);
+        if (pkt_mgr_schema->topic_id[i] < 0)
         {
-            PACKET_MANAGER_LOG_ERROR("failed to create topic");
+            PACKET_MANAGER_LOG_ERROR("failed to create topic %s", packet_stage_to_str(i));
             goto error_out;
         }
     }
 
-    return schema;
+    return pkt_mgr_schema;
 
 error_out:
-    packet_manager_schema_free(schema);
+    packet_manager_schema_free(pkt_mgr_schema);
     return NULL;
 }
 
+int packet_manager_schema_add_subscriber(struct packet_manager_schema *pkt_mgr_schema, enum packet_stage stage, on_packet_stage_callback cb, void *args)
+{
+    return mq_schema_subscribe(pkt_mgr_schema->mq, pkt_mgr_schema->topic_id[stage], (on_msg_cb_func *)cb, args);
+}
+
 /******************************************************************************
  * packet manager runtime
  ******************************************************************************/
@@ -169,10 +192,6 @@ static void packet_manager_runtime_free(struct packet_manager_runtime *runtime)
 {
     if (runtime)
     {
-        if (runtime->mq)
-        {
-            mq_runtime_free(runtime->mq);
-        }
         for (int i = 0; i < PACKET_STAGE_MAX; i++)
         {
             struct packet *pkt = NULL;
@@ -189,7 +208,7 @@ static void packet_manager_runtime_free(struct packet_manager_runtime *runtime)
     runtime = NULL;
 }
 
-static struct packet_manager_runtime *packet_manager_runtime_new(struct packet_manager_schema *schema)
+static struct packet_manager_runtime *packet_manager_runtime_new(uint16_t idx)
 {
     struct packet_manager_runtime *runtime = calloc(1, sizeof(struct packet_manager_runtime));
     if (runtime == NULL)
@@ -198,13 +217,7 @@ static struct packet_manager_runtime *packet_manager_runtime_new(struct packet_m
         return NULL;
     }
 
-    runtime->mq = mq_runtime_new(schema->mq);
-    if (runtime->mq == NULL)
-    {
-        PACKET_MANAGER_LOG_ERROR("failed to create mq_runtime");
-        free(runtime);
-        return NULL;
-    }
+    runtime->idx = idx;
 
     for (int i = 0; i < PACKET_STAGE_MAX; i++)
     {
@@ -216,19 +229,94 @@ static struct packet_manager_runtime *packet_manager_runtime_new(struct packet_m
 
 static void packet_manager_runtime_stat(struct packet_manager_runtime *runtime)
 {
-    PACKET_MANAGER_LOG_INFO("input_pkts: %lu, output_pkts: %lu, take_pkts: %lu, schedule_pkts: %lu",
-                            runtime->stat.input_pkts, runtime->stat.output_pkts, runtime->stat.take_pkts, runtime->stat.schedule_pkts);
+    PACKET_MANAGER_LOG_DEBUG("runtime[%d] => input_pkts: %lu, output_pkts: %lu, take_pkts: %lu, schedule_pkts: %lu, queue_len: {pre_routing: %lu, input: %lu, forward: %lu, output: %lu, post_routing: %lu}",
+                             runtime->idx,
+                             runtime->stat.input_pkts, runtime->stat.output_pkts,
+                             runtime->stat.take_pkts, runtime->stat.schedule_pkts,
+                             runtime->stat.curr_queue_len[PACKET_STAGE_PREROUTING],
+                             runtime->stat.curr_queue_len[PACKET_STAGE_INPUT],
+                             runtime->stat.curr_queue_len[PACKET_STAGE_FORWARD],
+                             runtime->stat.curr_queue_len[PACKET_STAGE_OUTPUT],
+                             runtime->stat.curr_queue_len[PACKET_STAGE_POSTROUTING]);
+}
+
+void packet_manager_runtime_init(struct packet_manager_runtime *pkt_mgr_rt, struct mq_runtime *mq_rt)
+{
+    pkt_mgr_rt->mq = mq_rt;
+}
+
+void packet_manager_runtime_input(struct packet_manager_runtime *pkt_mgr_rt, struct packet *pkt)
+{
+    pkt_mgr_rt->stat.input_pkts++;
+    pkt_mgr_rt->stat.curr_queue_len[PACKET_STAGE_PREROUTING]++;
+    TAILQ_INSERT_TAIL(&pkt_mgr_rt->queue[PACKET_STAGE_PREROUTING], pkt, stage_tqe);
+}
+
+struct packet *packet_manager_runtime_output(struct packet_manager_runtime *pkt_mgr_rt)
+{
+    struct packet *pkt = TAILQ_FIRST(&pkt_mgr_rt->queue[PACKET_STAGE_POSTROUTING]);
+    if (pkt)
+    {
+        pkt_mgr_rt->stat.output_pkts++;
+        pkt_mgr_rt->stat.curr_queue_len[PACKET_STAGE_POSTROUTING]--;
+        TAILQ_REMOVE(&pkt_mgr_rt->queue[PACKET_STAGE_POSTROUTING], pkt, stage_tqe);
+    }
+    return pkt;
+}
+
+void packet_manager_runtime_dispatch(struct packet_manager_runtime *pkt_mgr_rt)
+{
     for (int i = 0; i < PACKET_STAGE_MAX; i++)
     {
-        PACKET_MANAGER_LOG_INFO("curr_queue_len[%d]: %lu", i, runtime->stat.curr_queue_len[i]);
+        // packet_manager_runtime_stat(pkt_mgr_rt);
+
+        struct packet *pkt = NULL;
+        while ((pkt = TAILQ_FIRST(&pkt_mgr_rt->queue[i])))
+        {
+            TAILQ_REMOVE(&pkt_mgr_rt->queue[i], pkt, stage_tqe);
+            pkt_mgr_rt->stat.curr_queue_len[i]--;
+
+            mq_runtime_publish_message(pkt_mgr_rt->mq, i, pkt);
+            mq_runtime_dispatch(pkt_mgr_rt->mq);
+
+            if (packet_is_stolen(pkt))
+            {
+                continue;
+            }
+
+            if (i + 1 == PACKET_STAGE_MAX)
+            {
+                TAILQ_INSERT_TAIL(&pkt_mgr_rt->queue[i], pkt, stage_tqe);
+                pkt_mgr_rt->stat.curr_queue_len[i]++;
+                break;
+            }
+            else
+            {
+                TAILQ_INSERT_TAIL(&pkt_mgr_rt->queue[i + 1], pkt, stage_tqe);
+                pkt_mgr_rt->stat.curr_queue_len[i + 1]++;
+            }
+        }
     }
 }
 
+void packet_manager_runtime_take_packet(struct packet_manager_runtime *pkt_mgr_rt, struct packet *pkt)
+{
+    pkt_mgr_rt->stat.take_pkts++;
+    packet_set_stolen(pkt, true);
+}
+
+void packet_manager_runtime_schedule_packet(struct packet_manager_runtime *pkt_mgr_rt, struct packet *pkt, enum packet_stage stage)
+{
+    pkt_mgr_rt->stat.schedule_pkts++;
+    packet_set_stolen(pkt, false);
+    TAILQ_INSERT_TAIL(&pkt_mgr_rt->queue[stage], pkt, stage_tqe);
+}
+
 /******************************************************************************
- * Public API
+ *  packet manager
  ******************************************************************************/
 
-struct packet_manager *packet_manager_new(const char *toml_file)
+struct packet_manager *packet_manager_new(struct mq_schema *mq, const char *toml_file)
 {
     struct packet_manager *pkt_mgr = calloc(1, sizeof(struct packet_manager));
     if (pkt_mgr == NULL)
@@ -244,7 +332,7 @@ struct packet_manager *packet_manager_new(const char *toml_file)
         goto error_out;
     }
 
-    pkt_mgr->schema = packet_manager_schema_new();
+    pkt_mgr->schema = packet_manager_schema_new(mq);
     if (pkt_mgr->schema == NULL)
     {
         PACKET_MANAGER_LOG_ERROR("failed to create packet_manager_schema");
@@ -253,7 +341,7 @@ struct packet_manager *packet_manager_new(const char *toml_file)
 
     for (uint16_t i = 0; i < pkt_mgr->cfg->nr_worker_thread; i++)
     {
-        pkt_mgr->runtime[i] = packet_manager_runtime_new(pkt_mgr->schema);
+        pkt_mgr->runtime[i] = packet_manager_runtime_new(i);
         if (pkt_mgr->runtime[i] == NULL)
         {
             PACKET_MANAGER_LOG_ERROR("failed to create packet_manager_runtime");
@@ -272,73 +360,23 @@ void packet_manager_free(struct packet_manager *pkt_mgr)
 {
     if (pkt_mgr)
     {
-        packet_manager_config_free(pkt_mgr->cfg);
-        packet_manager_schema_free(pkt_mgr->schema);
-
         for (uint16_t i = 0; i < pkt_mgr->cfg->nr_worker_thread; i++)
         {
             if (pkt_mgr->runtime[i])
             {
-                PACKET_MANAGER_LOG_INFO("worker thread %d packet manager runtime stat", i);
                 packet_manager_runtime_stat(pkt_mgr->runtime[i]);
                 packet_manager_runtime_free(pkt_mgr->runtime[i]);
             }
         }
 
+        packet_manager_schema_free(pkt_mgr->schema);
+        packet_manager_config_free(pkt_mgr->cfg);
+
         free(pkt_mgr);
         pkt_mgr = NULL;
     }
 }
 
-void packet_manager_runtime_input(struct packet_manager_runtime *pkt_mgr_rt, struct packet *pkt)
-{
-    pkt_mgr_rt->stat.input_pkts++;
-    pkt_mgr_rt->stat.curr_queue_len[PACKET_STAGE_PREROUTING]++;
-    TAILQ_INSERT_TAIL(&pkt_mgr_rt->queue[PACKET_STAGE_PREROUTING], pkt, stage_tqe);
-}
-
-struct packet *packet_manager_runtime_output(struct packet_manager_runtime *pkt_mgr_rt)
-{
-    struct packet *pkt = TAILQ_FIRST(&pkt_mgr_rt->queue[PACKET_STAGE_POSTROUTING]);
-    if (pkt)
-    {
-        pkt_mgr_rt->stat.output_pkts++;
-        pkt_mgr_rt->stat.curr_queue_len[PACKET_STAGE_POSTROUTING]--;
-        TAILQ_REMOVE(&pkt_mgr_rt->queue[PACKET_STAGE_POSTROUTING], pkt, stage_tqe);
-    }
-    return pkt;
-}
-
-void packet_manager_runtime_dispatch(struct packet_manager_runtime *pkt_mgr_rt)
-{
-    for (int i = 0; i < PACKET_STAGE_MAX; i++)
-    {
-        struct packet *pkt = NULL;
-        while ((pkt = TAILQ_FIRST(&pkt_mgr_rt->queue[i])))
-        {
-            pkt_mgr_rt->stat.curr_queue_len[i]--;
-            TAILQ_REMOVE(&pkt_mgr_rt->queue[i], pkt, stage_tqe);
-            mq_runtime_publish_message(pkt_mgr_rt->mq, i, pkt);
-            mq_runtime_dispatch(pkt_mgr_rt->mq);
-
-            if (packet_is_stolen(pkt))
-            {
-                continue;
-            }
-
-            if (i + 1 == PACKET_STAGE_MAX)
-            {
-                break;
-            }
-            else
-            {
-                pkt_mgr_rt->stat.curr_queue_len[i + 1]++;
-                TAILQ_INSERT_TAIL(&pkt_mgr_rt->queue[i + 1], pkt, stage_tqe);
-            }
-        }
-    }
-}
-
 struct packet_manager_schema *packet_manager_get_schema(struct packet_manager *pkt_mgr)
 {
     if (pkt_mgr)
@@ -362,21 +400,3 @@ struct packet_manager_runtime *packet_manager_get_runtime(struct packet_manager
         return NULL;
     }
 }
-
-int packet_manager_schema_add_subscriber(struct packet_manager_schema *schema, enum packet_stage stage, on_packet_stage_callback cb, void *args)
-{
-    return mq_schema_subscribe(schema->mq, schema->topic_id[stage], (on_msg_cb_func *)cb, args);
-}
-
-void packet_manager_runtime_take_packet(struct packet_manager_runtime *pkt_mgr_rt, struct packet *pkt)
-{
-    pkt_mgr_rt->stat.take_pkts++;
-    packet_set_stolen(pkt, true);
-}
-
-void packet_manager_runtime_schedule_packet(struct packet_manager_runtime *pkt_mgr_rt, struct packet *pkt, enum packet_stage stage)
-{
-    pkt_mgr_rt->stat.schedule_pkts++;
-    packet_set_stolen(pkt, false);
-    TAILQ_INSERT_TAIL(&pkt_mgr_rt->queue[stage], pkt, stage_tqe);
-}
\ No newline at end of file
diff --git a/infra/packet_manager/packet_manager_private.h b/infra/packet_manager/packet_manager_private.h
index 85c5106..3f0dcf3 100644
--- a/infra/packet_manager/packet_manager_private.h
+++ b/infra/packet_manager/packet_manager_private.h
@@ -7,12 +7,15 @@ extern "C"
 
 #include "stellar/packet_manager.h"
 
-struct packet_manager *packet_manager_new(const char *toml_file);
+struct packet_manager *packet_manager_new(struct mq_schema *mq_schema, const char *toml_file);
 void packet_manager_free(struct packet_manager *pkt_mgr);
 
+void packet_manager_runtime_init(struct packet_manager_runtime *pkt_mgr_rt, struct mq_runtime *mq_rt);
 void packet_manager_runtime_input(struct packet_manager_runtime *pkt_mgr_rt, struct packet *pkt);
 struct packet *packet_manager_runtime_output(struct packet_manager_runtime *pkt_mgr_rt);
 void packet_manager_runtime_dispatch(struct packet_manager_runtime *pkt_mgr_rt);
+// for debug
+const char *packet_stage_to_str(enum packet_stage stage);
 
 #ifdef __cplusplus
 }
diff --git a/infra/packet_manager/test/CMakeLists.txt b/infra/packet_manager/test/CMakeLists.txt
index 3d12184..0c8314c 100644
--- a/infra/packet_manager/test/CMakeLists.txt
+++ b/infra/packet_manager/test/CMakeLists.txt
@@ -55,6 +55,9 @@ target_link_libraries(gtest_packet_filter packet_manager gtest)
 add_executable(gtest_packet_ldbc gtest_packet_ldbc.cpp)
 target_link_libraries(gtest_packet_ldbc packet_manager gtest)
 
+add_executable(gtest_packet_manager gtest_packet_manager.cpp)
+target_link_libraries(gtest_packet_manager packet_manager gtest)
+
 include(GoogleTest)
 gtest_discover_tests(gtest_tunnel)
 gtest_discover_tests(gtest_udp_utils)
@@ -74,4 +77,7 @@ gtest_discover_tests(gtest_packet_frag)
 gtest_discover_tests(gtest_packet_parser)
 gtest_discover_tests(gtest_packet_builder)
 gtest_discover_tests(gtest_packet_filter)
-gtest_discover_tests(gtest_packet_ldbc)
\ No newline at end of file
+gtest_discover_tests(gtest_packet_ldbc)
+gtest_discover_tests(gtest_packet_manager)
+
+file(COPY ../../../conf/ DESTINATION ./conf/)
\ No newline at end of file
diff --git a/infra/packet_manager/test/gtest_packet_manager.cpp b/infra/packet_manager/test/gtest_packet_manager.cpp
new file mode 100644
index 0000000..2fee4a4
--- /dev/null
+++ b/infra/packet_manager/test/gtest_packet_manager.cpp
@@ -0,0 +1,167 @@
+#include <gtest/gtest.h>
+
+#include "stellar/mq.h"
+#include "packet_parser.h"
+#include "packet_private.h"
+#include "packet_manager_private.h"
+
+/******************************************************************************
+ * [Protocols in frame: eth:ethertype:ip:ipv6:tcp]
+ ******************************************************************************
+ *
+ * Frame 1: 106 bytes on wire (848 bits), 106 bytes captured (848 bits)
+ * Ethernet II, Src: JuniperN_45:88:29 (2c:6b:f5:45:88:29), Dst: JuniperN_2a:a2:00 (5c:5e:ab:2a:a2:00)
+ * 	Destination: JuniperN_2a:a2:00 (5c:5e:ab:2a:a2:00)
+ * 	Source: JuniperN_45:88:29 (2c:6b:f5:45:88:29)
+ * 	Type: IPv4 (0x0800)
+ * Internet Protocol Version 4, Src: 210.77.88.163, Dst: 59.66.4.50
+ * 	0100 .... = Version: 4
+ * 	.... 0101 = Header Length: 20 bytes (5)
+ * 	Differentiated Services Field: 0x00 (DSCP: CS0, ECN: Not-ECT)
+ * 	Total Length: 92
+ * 	Identification: 0x0b4d (2893)
+ * 	000. .... = Flags: 0x0
+ * 	...0 0000 0000 0000 = Fragment Offset: 0
+ * 	Time to Live: 59
+ * 	Protocol: IPv6 (41)
+ * 	Header Checksum: 0x09c8 [validation disabled]
+ * 	[Header checksum status: Unverified]
+ * 	Source Address: 210.77.88.163
+ * 	Destination Address: 59.66.4.50
+ * Internet Protocol Version 6, Src: 2001:da8:200:900e:200:5efe:d24d:58a3, Dst: 2600:140e:6::1702:1058
+ * 	0110 .... = Version: 6
+ * 	.... 0000 0000 .... .... .... .... .... = Traffic Class: 0x00 (DSCP: CS0, ECN: Not-ECT)
+ * 	.... 0000 0000 0000 0000 0000 = Flow Label: 0x00000
+ * 	Payload Length: 32
+ * 	Next Header: TCP (6)
+ * 	Hop Limit: 64
+ * 	Source Address: 2001:da8:200:900e:200:5efe:d24d:58a3
+ * 	Destination Address: 2600:140e:6::1702:1058
+ * 	[Source ISATAP IPv4: 210.77.88.163]
+ * Transmission Control Protocol, Src Port: 52556, Dst Port: 80, Seq: 0, Len: 0
+ * 	Source Port: 52556
+ * 	Destination Port: 80
+ * 	[Stream index: 0]
+ * 	[Conversation completeness: Complete, WITH_DATA (31)]
+ * 	[TCP Segment Len: 0]
+ * 	Sequence Number: 0    (relative sequence number)
+ * 	Sequence Number (raw): 2172673142
+ * 	[Next Sequence Number: 1    (relative sequence number)]
+ * 	Acknowledgment Number: 0
+ * 	Acknowledgment number (raw): 0
+ * 	1000 .... = Header Length: 32 bytes (8)
+ * 	Flags: 0x002 (SYN)
+ * 	Window: 8192
+ * 	[Calculated window size: 8192]
+ * 	Checksum: 0xf757 [unverified]
+ * 	[Checksum Status: Unverified]
+ * 	Urgent Pointer: 0
+ * 	Options: (12 bytes), Maximum segment size, No-Operation (NOP), Window scale, No-Operation (NOP), No-Operation (NOP), SACK permitted
+ * 	[Timestamps]
+ */
+
+unsigned char data[] = {
+	0x5c, 0x5e, 0xab, 0x2a, 0xa2, 0x00, 0x2c, 0x6b, 0xf5, 0x45, 0x88, 0x29, 0x08, 0x00, 0x45, 0x00, 0x00, 0x5c, 0x0b, 0x4d, 0x00, 0x00, 0x3b, 0x29, 0x09, 0xc8,
+	0xd2, 0x4d, 0x58, 0xa3, 0x3b, 0x42, 0x04, 0x32, 0x60, 0x00, 0x00, 0x00, 0x00, 0x20, 0x06, 0x40, 0x20, 0x01, 0x0d, 0xa8, 0x02, 0x00, 0x90, 0x0e, 0x02, 0x00,
+	0x5e, 0xfe, 0xd2, 0x4d, 0x58, 0xa3, 0x26, 0x00, 0x14, 0x0e, 0x00, 0x06, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x17, 0x02, 0x10, 0x58, 0xcd, 0x4c, 0x00, 0x50,
+	0x81, 0x80, 0x5c, 0x76, 0x00, 0x00, 0x00, 0x00, 0x80, 0x02, 0x20, 0x00, 0xf7, 0x57, 0x00, 0x00, 0x02, 0x04, 0x04, 0xc4, 0x01, 0x03, 0x03, 0x08, 0x01, 0x01,
+	0x04, 0x02};
+
+#if 1
+TEST(PACKET_MANAGER, NEW_FREE)
+{
+	struct mq_schema *mq_schema = mq_schema_new();
+	EXPECT_TRUE(mq_schema);
+
+	struct packet_manager *pkt_mgr = packet_manager_new(mq_schema, "./conf/stellar.toml");
+	EXPECT_TRUE(pkt_mgr);
+
+	EXPECT_TRUE(packet_manager_get_schema(pkt_mgr));
+	EXPECT_TRUE(packet_manager_get_runtime(pkt_mgr, 0));
+	EXPECT_TRUE(packet_manager_get_runtime(pkt_mgr, 1) == NULL);
+
+	packet_manager_free(pkt_mgr);
+
+	mq_schema_free(mq_schema);
+}
+#endif
+
+static void on_packet_stage(enum packet_stage stage, struct packet *pkt, void *args)
+{
+	printf("stage: %s\n", packet_stage_to_str(stage));
+
+	static int count = 0;
+	EXPECT_TRUE(count == stage);
+	count++;
+
+	EXPECT_TRUE(packet_is_ctrl(pkt));
+	EXPECT_TRUE(args == NULL);
+}
+
+#if 1
+TEST(PACKET_MANAGER, NORNMAL)
+{
+	// global init
+	struct mq_schema *mq_schema = mq_schema_new();
+	EXPECT_TRUE(mq_schema);
+	struct mq_runtime *mq_rt = mq_runtime_new(mq_schema);
+	EXPECT_TRUE(mq_rt);
+
+	// module init
+	struct packet_manager *pkt_mgr = packet_manager_new(mq_schema, "./conf/stellar.toml");
+	EXPECT_TRUE(pkt_mgr);
+	struct packet_manager_schema *schema = packet_manager_get_schema(pkt_mgr);
+	EXPECT_TRUE(schema);
+	EXPECT_TRUE(packet_manager_schema_add_subscriber(schema, PACKET_STAGE_PREROUTING, on_packet_stage, NULL) == 0);
+	EXPECT_TRUE(packet_manager_schema_add_subscriber(schema, PACKET_STAGE_INPUT, on_packet_stage, NULL) == 0);
+	EXPECT_TRUE(packet_manager_schema_add_subscriber(schema, PACKET_STAGE_FORWARD, on_packet_stage, NULL) == 0);
+	EXPECT_TRUE(packet_manager_schema_add_subscriber(schema, PACKET_STAGE_OUTPUT, on_packet_stage, NULL) == 0);
+	EXPECT_TRUE(packet_manager_schema_add_subscriber(schema, PACKET_STAGE_POSTROUTING, on_packet_stage, NULL) == 0);
+
+	// per-thread init
+	struct packet_manager_runtime *runtime = packet_manager_get_runtime(pkt_mgr, 0);
+	EXPECT_TRUE(runtime);
+	packet_manager_runtime_init(runtime, mq_rt);
+
+	// per-thread run
+	struct packet pkt;
+	memset(&pkt, 0, sizeof(pkt));
+	packet_parse(&pkt, (const char *)data, sizeof(data));
+	packet_set_ctrl(&pkt, true);
+
+	packet_manager_runtime_input(runtime, &pkt);
+	packet_manager_runtime_dispatch(runtime);
+	EXPECT_TRUE(packet_manager_runtime_output(runtime) == &pkt);
+
+	// per-thread free
+
+	// module free
+	packet_manager_free(pkt_mgr);
+
+	// global free
+	mq_runtime_free(mq_rt);
+	mq_schema_free(mq_schema);
+}
+#endif
+
+#if 1
+TEST(PACKET_MANAGER, TAKE)
+{
+	// TODO
+	// packet_manager_runtime_take_packet(struct packet_manager_runtime *pkt_mgr_rt, struct packet *pkt);
+}
+#endif
+
+#if 1
+TEST(PACKET_MANAGER, SCHEDULE)
+{
+	// TODO
+	// packet_manager_runtime_schedule_packet(struct packet_manager_runtime *pkt_mgr_rt, struct packet *pkt, enum packet_stage stage);
+}
+#endif
+
+int main(int argc, char **argv)
+{
+	::testing::InitGoogleTest(&argc, argv);
+	return RUN_ALL_TESTS();
+}