feature: packet IO support IP reassembly

1 files changed, 27 insertions, 43 deletions

diff --git a/infra/stellar_core.c b/infra/stellar_core.c
index 8c6e704..cafc537 100644
--- a/infra/stellar_core.c
+++ b/infra/stellar_core.c
@@ -2,11 +2,14 @@
 #include <pthread.h>
 #include <sys/prctl.h>
 
+#include "stellar/stellar.h"
+#include "stellar/module_manager.h"
+
 #include "packet_io.h"
+#include "log_internal.h"
 #include "packet_internal.h"
+#include "utils_internal.h"
 #include "packet_manager_internal.h"
-
-#include "stellar/stellar.h"
 #include "module_manager_interna.h"
 
 #define CORE_LOG_FATAL(format, ...) STELLAR_LOG_FATAL(__thread_local_logger, "core", format, ##__VA_ARGS__)
@@ -19,7 +22,7 @@ static __attribute__((__used__)) const char *version = STELLAR_GIT_VERSION;
 static __attribute__((__used__)) const char *version = "Unknown";
 #endif
 
-struct stellar_thread
+struct thread
 {
     pthread_t tid;
     uint16_t idx;
@@ -35,16 +38,16 @@ struct stellar
     struct packet_io *pkt_io;
     struct mq_schema *mq_schema;
     struct stellar_module_manager *mod_mgr;
-    struct stellar_thread threads[MAX_THREAD_NUM];
+    struct thread threads[MAX_THREAD_NUM];
 };
 
 static void *worker_thread(void *arg)
 {
-    int nr_pkt_rcv = 0;
+    int nr_recv = 0;
     char thread_name[16] = {0};
     struct packet *pkt = NULL;
-    struct packet packets[RX_BURST_MAX];
-    struct stellar_thread *thread = (struct stellar_thread *)arg;
+    struct packet *pkts[RX_BURST_MAX] = {NULL};
+    struct thread *thread = (struct thread *)arg;
     uint16_t thread_id = thread->idx;
     struct stellar *st = thread->st;
     struct packet_io *pkt_io = st->pkt_io;
@@ -58,63 +61,44 @@ static void *worker_thread(void *arg)
     __thread_local_logger = st->logger;
     stellar_module_manager_register_thread(mod_mgr, thread_id, mq_rt);
 
-    if (packet_io_init(pkt_io, thread_id) != 0)
+    if (packet_manager_init(pkt_mgr, thread_id, mq_rt) != 0)
     {
-        CORE_LOG_ERROR("unable to init packet io");
+        CORE_LOG_ERROR("unable to init packet manager");
         return NULL;
     }
-    packet_manager_init(pkt_mgr, thread_id, mq_rt);
 
     ATOMIC_SET(&thread->is_runing, 1);
     CORE_LOG_FATAL("worker thread %d runing", thread_id);
 
     while (ATOMIC_READ(&st->need_exit) == 0)
     {
-        // TODO
-        memset(packets, 0, sizeof(packets));
-        nr_pkt_rcv = packet_io_ingress(pkt_io, thread_id, packets, RX_BURST_MAX);
-        if (nr_pkt_rcv == 0)
+        nr_recv = packet_io_recv(pkt_io, thread_id, pkts, RX_BURST_MAX);
+        for (int i = 0; i < nr_recv; i++)
         {
-            goto idle_tasks;
+            packet_manager_ingress(pkt_mgr, thread_id, pkts[i]);
         }
 
-        for (int i = 0; i < nr_pkt_rcv; i++)
+        packet_manager_dispatch(pkt_mgr, thread_id);
+        while ((pkt = packet_manager_egress(pkt_mgr, thread_id)))
         {
-            // TODO alloc struct packet from packet pool
-            pkt = calloc(1, sizeof(struct packet));
-            memcpy(pkt, &packets[i], sizeof(struct packet));
-            pkt->need_free = 1;
-
-            packet_manager_ingress(pkt_mgr, thread_id, pkt);
-            packet_manager_dispatch(pkt_mgr, thread_id);
-            pkt = packet_manager_egress(pkt_mgr, thread_id);
-
-            if (pkt == NULL)
-            {
-                continue;
-            }
-
             if (packet_get_action(pkt) == PACKET_ACTION_DROP)
             {
-                packet_io_drop(pkt_io, thread_id, pkt, 1);
-                packet_free(pkt);
+                packet_io_drop(pkt_io, thread_id, &pkt, 1);
             }
             else
             {
-                packet_io_egress(pkt_io, thread_id, pkt, 1);
-                packet_free(pkt);
+                packet_io_send(pkt_io, thread_id, &pkt, 1);
             }
-            stellar_polling_dispatch(mod_mgr);
         }
 
-    idle_tasks:
         stellar_polling_dispatch(mod_mgr);
-
-        if (nr_pkt_rcv == 0)
+        packet_io_polling(pkt_io, thread_id);
+        if (nr_recv == 0)
         {
             packet_io_yield(pkt_io, thread_id);
         }
     }
+
     stellar_module_manager_unregister_thread(mod_mgr, thread_id);
     mq_runtime_free(mq_rt);
 
@@ -126,9 +110,9 @@ static void *worker_thread(void *arg)
 
 static int stellar_thread_run(struct stellar *st)
 {
-    for (uint16_t i = 0; i < st->thread_num; i++)
+    for (uint64_t i = 0; i < st->thread_num; i++)
     {
-        struct stellar_thread *thread = &st->threads[i];
+        struct thread *thread = &st->threads[i];
         thread->idx = i;
         thread->is_runing = 0;
         thread->st = st;
@@ -144,14 +128,14 @@ static int stellar_thread_run(struct stellar *st)
 
 static void stellar_thread_join(struct stellar *st)
 {
-    for (uint16_t i = 0; i < st->thread_num; i++)
+    for (uint64_t i = 0; i < st->thread_num; i++)
     {
         if (st->threads[i].is_runing == 0)
         {
             continue;
         }
 
-        struct stellar_thread *thread = &st->threads[i];
+        struct thread *thread = &st->threads[i];
         pthread_join(thread->tid, NULL);
     }
 }
@@ -180,7 +164,7 @@ struct stellar *stellar_new(const char *toml_file)
     __thread_local_logger = st->logger;
     CORE_LOG_FATAL("stellar start (version: %s)", version);
 
-    if (load_and_validate_toml_integer_config(toml_file, "packet_io.nr_worker_thread", (uint64_t *)&st->thread_num, 1, MAX_THREAD_NUM) != 0)
+    if (load_toml_integer_config(toml_file, "packet_io.thread_num", (uint64_t *)&st->thread_num, 1, MAX_THREAD_NUM) != 0)
     {
         CORE_LOG_ERROR("unable to get thread number from config file");
         goto error_out;