feat: added in worker dpdk on-off filltring

worker/include/dpdk_filter/proc_packets.h

@@ -20,6 +20,6 @@ bool check_is_exception(uint16_t *port);
 void pakage_processing(struct net_port *port_in, struct net_port *port_out,
                        struct net_port *port_exception, uint16_t queue_number,
                        uint16_t nb_pkts, struct rte_mbuf **pkts,
-                       struct BASE_POLICY *policy);
+                       struct BASE_POLICY *policy, bool filtring_is_turned_off);
 
 #endif

worker/src/dpdk_filter/ip_cache.c

@@ -243,7 +243,7 @@ static int insert_ip_main_record(const char *ip_str,
 }
 
 static int delete_ip_categories(const char *ip_str) {
-  const char *sql = "DELETE FROM ip_categories_table WHERE ip_str = ?";
+  const char *sql = "DELETE FROM categories_table WHERE ip_str = ?";
   sqlite3_stmt *stmt = NULL;
   int ret = sqlite3_prepare_v2(ip_cache_table, sql, -1, &stmt, NULL);
   if (ret != SQLITE_OK) {
@@ -260,7 +260,7 @@ static int delete_ip_categories(const char *ip_str) {
 
 static int insert_ip_categories(const char *ip_str,
                                 struct node_cache_ip *node) {
-  const char *sql = "INSERT INTO ip_categories_table (ip_str, "
+  const char *sql = "INSERT INTO categories_table (ip_str, "
                     "certain_category) VALUES (?, ?)";
   sqlite3_stmt *stmt = NULL;
   int ret = sqlite3_prepare_v2(ip_cache_table, sql, -1, &stmt, NULL);

worker/src/dpdk_filter/main.c

@@ -111,7 +111,7 @@ int main(int argc, char **argv) {
   while (running) {
     forward_to_out(port_exception, port_in, queue_number);
     pakage_processing(port_in, port_out, port_exception, queue_number, nb_pkts,
-                      pkts, &policy);
+                      pkts, &policy, false);
     forward_to_out(port_out, port_in, queue_number);
 
     if (++timer_check_counter >= timer_check_interval) {

worker/src/dpdk_filter/proc_packets.c

@@ -1,6 +1,7 @@
 #include "proc_packets.h"
 #include "domain_cache.h"
 #include "ip_cache.h"
+#include <stdatomic.h>
 
 extern bool worker_classify(const char *type, const char *target,
                             struct requested_classification *out_req);
@@ -36,11 +37,19 @@ bool check_is_exception(uint16_t *port) {
 void pakage_processing(struct net_port *port_in, struct net_port *port_out,
                        struct net_port *port_exception, uint16_t queue_number,
                        uint16_t nb_pkts, struct rte_mbuf **pkts,
-                       struct BASE_POLICY *policy) {
+                       struct BASE_POLICY *policy,
+                       bool filtring_is_turned_off) {
 
   uint16_t nb_rx =
       rte_eth_rx_burst(port_in->port_id, queue_number, pkts, nb_pkts);
 
+  if (atomic_load(&filtring_is_turned_off)) {
+    for (int i = 0; i < nb_rx; i++) {
+      package_sending_decision(true, pkts[i], port_out, queue_number);
+    }
+    return;
+  }
+
   for (int i = 0; i < nb_rx; i++) {
 
     struct info_of_pakage info_pac;