Merge branch 'separate_swarmkv_priority_len_in_out' into 'rel'v3.1.38

separate in out direction for queue_len stored in swarmkv

See merge request tango/shaping-engine!98

2 files changed, 112 insertions, 6 deletions

diff --git a/shaping/test/gtest_shaper.cpp b/shaping/test/gtest_shaper.cpp
index 78b8247..b342eda 100644
--- a/shaping/test/gtest_shaper.cpp
+++ b/shaping/test/gtest_shaper.cpp
@@ -1199,6 +1199,94 @@ TEST(two_session_diff_priority_same_profile, profile_timer_test)
     stub_clear_matched_shaping_rules();
 }
 
+/*session1 match rule1; session2 match rule2
+  rule1:
+      priority:1
+      primary profile_a: (priority 1)
+  rule2:
+      priority:2
+      primary profile_a: (priority 2)
+
+profile_a: in limit 1000, out limit 1000
+*/
+TEST(two_session_diff_priority_same_profile, one_direction_dont_block_another)
+{
+    struct stub_pkt_queue expec_tx_queue1;
+    struct stub_pkt_queue expec_tx_queue2;
+    struct stub_pkt_queue *actual_tx_queue;
+    struct shaping_ctx *ctx = NULL;
+    struct shaping_flow *sf1 = NULL;
+    struct shaping_flow *sf2 = NULL;
+    long long rule_ids[] = {1, 2};
+    long long rule_id1[] = {1};
+    long long rule_id2[] = {2};
+    int profile_nums[] = {1, 1};
+    int prioritys[] = {1, 2};
+    int profile_id[][MAX_REF_PROFILE] = {{0}, {0}};
+    
+
+    TAILQ_INIT(&expec_tx_queue1);
+    TAILQ_INIT(&expec_tx_queue2);
+    stub_init();
+    
+    ctx = shaping_engine_init();
+    ASSERT_TRUE(ctx != NULL);
+    sf1 = shaping_flow_new(&ctx->thread_ctx[0]);
+    ASSERT_TRUE(sf1 != NULL);
+    sf2 = shaping_flow_new(&ctx->thread_ctx[1]);
+    ASSERT_TRUE(sf2 != NULL);
+
+    stub_set_matched_shaping_rules(2, rule_ids, prioritys, profile_nums, profile_id);
+
+    stub_set_token_bucket_avl_per_sec(0, 1000, SHAPING_DIR_OUT, PROFILE_LIMIT_DIRECTION_INCOMING_OUTGOING);
+    stub_set_token_bucket_avl_per_sec(0, 1000, SHAPING_DIR_IN, PROFILE_LIMIT_DIRECTION_INCOMING_OUTGOING);
+    actual_tx_queue = stub_get_tx_queue();
+    shaper_rules_update(&ctx->thread_ctx[0], sf1, rule_id1, 1);
+    shaper_rules_update(&ctx->thread_ctx[1], sf2, rule_id2, 1);
+
+    /*******send packets***********/
+    send_packets(&ctx->thread_ctx[0], sf1, 100, 100, SHAPING_DIR_OUT, &expec_tx_queue1, 1, 0);
+    send_packets(&ctx->thread_ctx[1], sf2, 100, 100, SHAPING_DIR_IN, &expec_tx_queue2, 1, 0);
+    ASSERT_EQ(0, judge_packet_eq(&expec_tx_queue1, actual_tx_queue, 10));
+    ASSERT_EQ(0, judge_packet_eq(&expec_tx_queue2, actual_tx_queue, 10));
+    ASSERT_TRUE(TAILQ_EMPTY(actual_tx_queue));
+
+    sleep(3);//wait profile timer to expire, to refresh priority queue_len to swarmkv
+    for (int i = 0; i < 500; i++) {
+        stub_curr_time_ns_inc(STUB_TIME_INC_FOR_PACKET);//inc time to refresh stat in timer
+    }
+    polling_entry(ctx->thread_ctx[0].sp, ctx->thread_ctx[0].stat, &ctx->thread_ctx[0]);//timer triggered in polling
+    polling_entry(ctx->thread_ctx[1].sp, ctx->thread_ctx[1].stat, &ctx->thread_ctx[1]);
+
+    stub_curr_time_s_inc(1);//inc time to refresh hmget interval
+    while (!TAILQ_EMPTY(&expec_tx_queue2)) {//线程0中优先级为1的session阻断OUT方向，线程1中的session优先级为2，但是IN方向不受影响
+        stub_refresh_token_bucket(0);
+        polling_entry(ctx->thread_ctx[1].sp, ctx->thread_ctx[1].stat, &ctx->thread_ctx[1]);//first polling request token
+        polling_entry(ctx->thread_ctx[1].sp, ctx->thread_ctx[1].stat, &ctx->thread_ctx[1]);
+        stub_curr_time_ns_inc(STUB_TIME_INC_FOR_PACKET);
+
+        ASSERT_EQ(0, judge_packet_eq(&expec_tx_queue2, actual_tx_queue, 1));
+    }
+
+    while (!TAILQ_EMPTY(&expec_tx_queue1)) {
+        stub_refresh_token_bucket(0);
+        polling_entry(ctx->thread_ctx[0].sp, ctx->thread_ctx[0].stat, &ctx->thread_ctx[0]);//first polling request token
+        polling_entry(ctx->thread_ctx[0].sp, ctx->thread_ctx[0].stat, &ctx->thread_ctx[0]);//then send pkt
+        stub_curr_time_ns_inc(STUB_TIME_INC_FOR_PACKET);
+
+        ASSERT_EQ(0, judge_packet_eq(&expec_tx_queue1, actual_tx_queue, 1));//sf1 priority 1
+    }
+
+    ASSERT_TRUE(TAILQ_EMPTY(actual_tx_queue));
+
+    shaping_flow_free(&ctx->thread_ctx[0], sf1);
+    shaping_flow_free(&ctx->thread_ctx[1], sf2);
+    fieldstat_global_disable_prometheus_endpoint();
+    shaper_thread_resource_clear();
+    shaping_engine_destroy(ctx);
+    stub_clear_matched_shaping_rules();
+}
+
 /*session1 match rule1 & rule2; session2 match rule3
   rule1:
       priority:1
diff --git a/shaping/test/stub.cpp b/shaping/test/stub.cpp
index 9285120..be3f29d 100644
--- a/shaping/test/stub.cpp
+++ b/shaping/test/stub.cpp
@@ -15,6 +15,7 @@
 #include <assert.h>
 #include <pthread.h>
 #include "shaper.h"
+#include "shaper_stat.h"
 #include "stub.h"
 #include "shaper_maat.h"
 #include "log.h"
@@ -51,7 +52,7 @@ static int pf_async_times[MAX_STUB_PROFILE_NUM];
 vector<struct stub_token_thread_arg> pf_async_thread[MAX_STUB_PROFILE_NUM];
 struct stub_matched_rules matched_rules;
 struct shaping_profile pf_array[MAX_STUB_PROFILE_NUM];
-static int profile_priority_len[MAX_STUB_PROFILE_NUM][10] = {{0}};
+static int profile_priority_len[MAX_STUB_PROFILE_NUM][SHAPING_PRIORITY_NUM_MAX][SHAPING_DIR_MAX];
 
 static unsigned long long curr_time_ns = 2000000000;//2s
 static unsigned int curr_time_s = 0;
@@ -204,6 +205,7 @@ void stub_init()
     TAILQ_INIT(&tx_queue);
     memset(&matched_rules, 0, sizeof(struct stub_matched_rules));
     memset(&pf_array, 0, MAX_STUB_PROFILE_NUM * sizeof(struct shaping_profile));
+    memset(&profile_priority_len, 0, MAX_STUB_PROFILE_NUM * SHAPING_PRIORITY_NUM_MAX * SHAPING_DIR_MAX * sizeof(int));
 
     for (i = 0; i < MAX_STUB_PROFILE_NUM; i++) {
         pf_curr_avl_token[i].in_limit_bandwidth = DEFAULT_AVALIABLE_TOKEN_PER_SEC;
@@ -316,10 +318,18 @@ static void swarmkv_hincrby_cmd_func(char *cmd_str, swarmkv_on_reply_callback_t
     int profile_id;
     int priority;
     int value;
+    char direction[5] = {0};
+    enum shaping_packet_dir dir;
     struct swarmkv_reply *reply = (struct swarmkv_reply*)calloc(1, sizeof(struct swarmkv_reply));
 
-    sscanf(cmd_str, "HINCRBY tsg-shaping-%d priority-%d %d", &profile_id, &priority, &value);
-    profile_priority_len[profile_id][priority] += value;
+    sscanf(cmd_str, "HINCRBY tsg-shaping-%d priority-%d-%s %d", &profile_id, &priority, direction, &value);
+    if (strncmp(direction, "in", 2) == 0) {
+        dir = SHAPING_DIR_IN;
+    } else {
+        dir = SHAPING_DIR_OUT;
+    }
+
+    profile_priority_len[profile_id][priority][dir] += value;
 
     reply->type = SWARMKV_REPLY_INTEGER;
     cb(reply, cb_arg);
@@ -335,12 +345,20 @@ static void swarmkv_hmget_cmd_func(char *cmd_str, swarmkv_on_reply_callback_t *
     int priority[10];
     int ret;
     int priority_num;
+    char direction[5] = {0};
+    enum shaping_packet_dir dir;
     struct swarmkv_reply *reply = (struct swarmkv_reply*)calloc(1, sizeof(struct swarmkv_reply));
 
-    ret = sscanf(cmd_str, "HMGET tsg-shaping-%d priority-%d priority-%d priority-%d priority-%d priority-%d priority-%d priority-%d priority-%d priority-%d", 
-                    &profile_id, &priority[0], &priority[1], &priority[2], &priority[3], &priority[4], &priority[5], &priority[6], &priority[7], &priority[8]);
+    ret = sscanf(cmd_str, "HMGET tsg-shaping-%d priority-%d-%s priority-%d-%*s priority-%d-%*s priority-%d-%*s priority-%d-%*s priority-%d-%*s priority-%d-%*s priority-%d-%*s priority-%d-%*s", 
+                    &profile_id, &priority[0], direction, &priority[1], &priority[2], &priority[3], &priority[4], &priority[5], &priority[6], &priority[7], &priority[8]);
     priority_num = ret - 1;
 
+    if (strncmp(direction, "in", 2) == 0) {
+        dir = SHAPING_DIR_IN;
+    } else {
+        dir = SHAPING_DIR_OUT;
+    }
+
     reply->type = SWARMKV_REPLY_ARRAY;
     reply->n_element = priority_num;
     reply->elements = (struct swarmkv_reply**)calloc(priority_num, sizeof(struct swarmkv_reply*));
@@ -349,7 +367,7 @@ static void swarmkv_hmget_cmd_func(char *cmd_str, swarmkv_on_reply_callback_t *
         reply->elements[i]->type = SWARMKV_REPLY_STRING;
 
         char tmp_str[128] = {0};
-        sprintf(tmp_str, "%d", profile_priority_len[profile_id][priority[i]]);
+        sprintf(tmp_str, "%d", profile_priority_len[profile_id][priority[i]][dir]);
         reply->elements[i]->str = (char *)calloc(1, strlen(tmp_str));
         memcpy(reply->elements[i]->str, tmp_str, strlen(tmp_str));
         reply->elements[i]->len = strlen(tmp_str);