bugfix: call duplicated packet filter too many times

conf/log.toml

@@ -1,4 +1,4 @@
 [log]
 output = file      # stderr, file
 file   = "log/stellar.log"
-level  = DEBUG       # TRACE, DEBUG, INFO, WARN, ERROR, FATAL
+level  = FATAL       # TRACE, DEBUG, INFO, WARN, ERROR, FATAL

conf/stellar.toml

@@ -19,24 +19,24 @@ bucket_num = 4096  # range: [1, 4294967295]
 
 [session_manager]
 # max session number
-max_tcp_session_num = 40960
+max_tcp_session_num = 50000
-max_udp_session_num = 40960
+max_udp_session_num = 50000
 
 # session overload evict
 tcp_overload_evict_old_sess = 1 # 1: evict old session, 0: bypass new session
 udp_overload_evict_old_sess = 1 # 1: evict old session, 0: bypass new session
 
 # TCP timeout
-tcp_init_timeout = 5000            # range: [1, 60000] (ms)
+tcp_init_timeout = 5000           # range: [1, 60000] (ms)
-tcp_handshake_timeout = 10000      # range: [1, 60000] (ms)
+tcp_handshake_timeout = 5000      # range: [1, 60000] (ms)
-tcp_data_timeout = 3600000         # range: [1, 15999999000] (ms)
+tcp_data_timeout = 5000           # range: [1, 15999999000] (ms)
-tcp_half_closed_timeout = 120000   # range: [1, 604800000] (ms)
+tcp_half_closed_timeout = 5000    # range: [1, 604800000] (ms)
-tcp_time_wait_timeout = 15000      # range: [1, 600000] (ms)
+tcp_time_wait_timeout = 5000      # range: [1, 600000] (ms)
-tcp_discard_timeout = 90000        # range: [1, 15999999000] (ms)
+tcp_discard_timeout = 10000       # range: [1, 15999999000] (ms)
-tcp_unverified_rst_timeout = 10000 # range: [1, 600000] (ms)
+tcp_unverified_rst_timeout = 5000 # range: [1, 600000] (ms)
 # UDP timeout
-udp_data_timeout = 10000    # range: [1, 15999999000] (ms)
+udp_data_timeout = 5000    # range: [1, 15999999000] (ms)
-udp_discard_timeout = 90000 # range: [1, 15999999000] (ms)
+udp_discard_timeout = 5000 # range: [1, 15999999000] (ms)
 
 # duplicate packet filter
 duplicated_packet_filter_enable = 1
@@ -53,4 +53,4 @@ evicted_session_filter_error_rate = 0.00001 # range: [0.0, 1.0]
 # TCP reassembly (Per direction)
 tcp_reassembly_enable = 1
 tcp_reassembly_max_timeout = 10000 # range: [1, 60000] (ms)
-tcp_reassembly_max_segments = 32   # range: [2, 32]
+tcp_reassembly_max_segments = 128  # range: [2, 512]

src/packet_io/marsio_io.cpp

@@ -22,7 +22,7 @@ struct marsio_io
  * Private API
  ******************************************************************************/
 
-static int is_keepalive_packet(const char *data, int len)
+static inline int is_keepalive_packet(const char *data, int len)
 {
     if (data == NULL || len < (int)(sizeof(struct ethhdr)))
     {

src/plugin/plugin_manager.cpp

@@ -16,10 +16,10 @@ void *plugin_manager_new_ctx(struct session *sess)
 void plugin_manager_free_ctx(void *ctx)
 {
     // TODO
-    struct session *sess = (struct session *)ctx;
+    // struct session *sess = (struct session *)ctx;
-    char buff[4096] = {0};
+    // char buff[4096] = {0};
-    session_to_json(sess, buff, sizeof(buff));
+    // session_to_json(sess, buff, sizeof(buff));
-    PLUGIN_MANAGER_LOG_DEBUG("=> SESSION : %s", buff);
+    // PLUGIN_MANAGER_LOG_DEBUG("=> SESSION : %s", buff);
 }
 
 struct plugin_manager *plugin_manager_new(void)

src/session/session_manager.cpp

@@ -211,9 +211,9 @@ static int check_options(const struct session_manager_options *opts)
                SESSION_LOG_ERROR("invalid tcp_reassembly_max_timeout: %u, supported range: [1, 60000]", opts->tcp_reassembly_max_timeout);
                return -1;
           }
-          if (opts->tcp_reassembly_max_segments < 2 || opts->tcp_reassembly_max_segments > 32)
+          if (opts->tcp_reassembly_max_segments < 2 || opts->tcp_reassembly_max_segments > 512)
           {
-               SESSION_LOG_ERROR("invalid tcp_reassembly_max_segments: %u, supported range: [2, 32]", opts->tcp_reassembly_max_segments);
+               SESSION_LOG_ERROR("invalid tcp_reassembly_max_segments: %u, supported range: [2, 512]", opts->tcp_reassembly_max_segments);
                return -1;
           }
      }
@@ -312,7 +312,7 @@ static void tcp_update(struct session_manager *mgr, struct session *sess, enum s
           return;
      }
 
-     if (flags & TH_SYN)
+     if (unlikely(flags & TH_SYN))
      {
           tcp_reassembly_set_recv_next(half->assembler, half->seq + 1);
      }
@@ -630,7 +630,7 @@ static struct session *session_manager_new_tcp_session(struct session_manager *m
      session_timer_update(mgr->sess_timer, sess, now + timeout);
      session_table_add(mgr->tcp_sess_table, key, sess);
 
-     if (session_get_stat(sess, dir, STAT_RAW_PKTS_RX) < 3 && mgr->opts.duplicated_packet_filter_enable)
+     if (mgr->opts.duplicated_packet_filter_enable)
      {
           duplicated_packet_filter_add(mgr->dup_pkt_filter, pkt, now);
      }
@@ -696,11 +696,6 @@ static int session_manager_update_tcp_session(struct session_manager *mgr, struc
      // update tcp
      tcp_update(mgr, sess, dir, tcp_layer, now);
 
-     if (mgr->opts.duplicated_packet_filter_enable)
-     {
-          duplicated_packet_filter_add(mgr->dup_pkt_filter, pkt, now);
-     }
-
      // set closing reason
      if (next_state == SESSION_STATE_CLOSING && !session_get_closing_reason(sess))
      {

src/session/session_manager.h

@@ -49,7 +49,7 @@ struct session_manager_options
     // TCP reassembly
     uint8_t tcp_reassembly_enable;
     uint32_t tcp_reassembly_max_timeout;  // range: [1, 60000] (ms)
-    uint32_t tcp_reassembly_max_segments; // range: [2, 32]
+    uint32_t tcp_reassembly_max_segments; // range: [2, 512]
 };
 
 struct session_manager_stat

src/session/session_private.h

@@ -15,7 +15,7 @@ extern "C"
 #include "tcp_reassembly.h"
 #include "session_manager.h"
 
-#define EX_DATA_MAX_COUNT 16
+#define EX_DATA_MAX_COUNT 4
 
 // tuple6 str format: "src_addr:src_port -> dst_addr:dst_port, proto: ip_proto, domain: domain"
 // tuple6 max len: 46 + 1 + 5 + 4 + 46 + 1 + 5 + 9 + 1 + 10 + 20 = 107
@@ -31,35 +31,39 @@ struct tcp_half
     uint8_t flags;
 };
 
-struct session
+/*
+ * sizeof(struct session) = 1024 bytes
+ * max thread number = 128
+ * per thread max tcp session number = 50000
+ * per thread max udp session number = 50000
+ *
+ * session memory usage = 128 * (50000 + 50000) * 1024 = 13107200000 bytes = 12.2 GB
+ */
+struct session // 1024 bytes
 {
-    int dup;
+    uint64_t id;                                   //   8 bytes
-    uint64_t id;
+    uint64_t stats[MAX_DIRECTION][MAX_STAT];       // 480 bytes
-    uint64_t stats[MAX_DIRECTION][MAX_STAT];
+    uint64_t timestamps[MAX_TIMESTAMP];            //  16 bytes
-    uint64_t timestamps[MAX_TIMESTAMP];
+    struct tcp_half tcp_halfs[MAX_DIRECTION];      //  80 bytes
-
+    struct timeout timeout;                        //  72 bytes -- used for timer
-    struct tuple6 tuple;
+    struct list_head lru;                          //  16 bytes -- used for lru queue
-    char tuple_str[TUPLE6_STR_SIZE];
+    struct list_head free;                         //  16 bytes -- used for free queue
-    enum session_direction tuple_dir;
+    struct list_head evicte;                       //  16 bytes -- used for evicte queue
-    enum session_direction cur_dir;
+    UT_hash_handle hh;                             //  56 bytes -- used for hash table
-    enum session_type type;
+    struct tuple6 tuple;                           //  56 bytes
-    enum session_state state;
+    char tuple_str[TUPLE6_STR_SIZE];               // 108 bytes
-    enum closing_reason reason;
+    const struct packet *first_pkt[MAX_DIRECTION]; //  16 bytes
-
+    const struct packet *curr_pkt;                 //   8 bytes
-    const struct packet *first_pkt[MAX_DIRECTION];
+    void *ex_data[EX_DATA_MAX_COUNT];              //  32 bytes
-    const struct packet *curr_pkt;
+    void *user_data;                               //   8 bytes
-
+    int is_symmetric;                              //   4 bytes
-    void *ex_data[EX_DATA_MAX_COUNT];
+    int dup;                                       //   4 bytes
-    void *user_data;
+    enum session_direction tuple_dir;              //   4 bytes
-    struct tcp_half tcp_halfs[MAX_DIRECTION];
+    enum session_direction cur_dir;                //   4 bytes
-
+    enum session_type type;                        //   4 bytes
-    struct timeout timeout;  // used for timer
+    enum session_state state;                      //   4 bytes
-    struct list_head lru;    // used for lru queue
+    enum closing_reason reason;                    //   4 bytes
-    struct list_head free;   // used for free queue
+    struct session_manager_stat *mgr_stat;         //   8 bytes
-    struct list_head evicte; // used for evicte queue
-    UT_hash_handle hh;       // used for hash table
-
-    struct session_manager_stat *mgr_stat;
 };
 
 void session_init(struct session *sess);

src/stellar/stellar.cpp

@@ -352,7 +352,7 @@ static inline void stellar_stat_output_cron(void *ctx)
 
 int main(int argc, char **argv)
 {
-    uint8_t nr_threads;
+    uint8_t nr_threads = 0;
     struct cron_task stat_task =
         {
             .callback = stellar_stat_output_cron,
@@ -381,6 +381,7 @@ int main(int argc, char **argv)
         goto error_out;
     }
     stellar_config_print(config);
+    STELLAR_LOG_DEBUG("sizeof(struct session) = %lu bytes", sizeof(struct session));
     nr_threads = config->io_opts.nr_threads;
 
     if (id_generator_init(config->dev_opts.base, config->dev_opts.offset) != 0)

src/tcp_reassembly/tcp_reassembly.cpp

@@ -95,6 +95,11 @@ void tcp_reassembly_free(struct tcp_reassembly *assembler)
 // return: -1: failed (no space)
 int tcp_reassembly_push(struct tcp_reassembly *assembler, struct tcp_segment *seg, uint64_t now)
 {
+    if (assembler == NULL)
+    {
+        return -1;
+    }
+
     if (assembler->cur_seg_num >= assembler->max_seg_num)
     {
         TCP_REASSEMBLY_LOG_ERROR("assembler is full");
@@ -121,9 +126,12 @@ int tcp_reassembly_push(struct tcp_reassembly *assembler, struct tcp_segment *se
 
 struct tcp_segment *tcp_reassembly_pop(struct tcp_reassembly *assembler)
 {
-    struct interval_tree_node *node;
+    if (assembler == NULL)
+    {
+        return NULL;
+    }
 
-    node = interval_tree_iter_first(&assembler->root, assembler->recv_next, assembler->recv_next);
+    struct interval_tree_node *node = interval_tree_iter_first(&assembler->root, assembler->recv_next, assembler->recv_next);
     if (node == NULL)
     {
         return NULL;
@@ -167,6 +175,11 @@ struct tcp_segment *tcp_reassembly_pop(struct tcp_reassembly *assembler)
 
 struct tcp_segment *tcp_reassembly_expire(struct tcp_reassembly *assembler, uint64_t now)
 {
+    if (assembler == NULL)
+    {
+        return NULL;
+    }
+
     if (list_empty(&assembler->list))
     {
         return NULL;
@@ -188,6 +201,11 @@ struct tcp_segment *tcp_reassembly_expire(struct tcp_reassembly *assembler, uint
 
 void tcp_reassembly_inc_recv_next(struct tcp_reassembly *assembler, uint32_t offset)
 {
+    if (assembler == NULL)
+    {
+        return;
+    }
+
     assembler->recv_next += offset;
     if (assembler->recv_next > UINT32_MAX)
     {
@@ -197,10 +215,20 @@ void tcp_reassembly_inc_recv_next(struct tcp_reassembly *assembler, uint32_t off
 
 void tcp_reassembly_set_recv_next(struct tcp_reassembly *assembler, uint32_t seq)
 {
+    if (assembler == NULL)
+    {
+        return;
+    }
+
     assembler->recv_next = seq;
 }
 
 uint32_t tcp_reassembly_get_recv_next(struct tcp_reassembly *assembler)
 {
+    if (assembler == NULL)
+    {
+        return 0;
+    }
+
     return assembler->recv_next;
 }