Merge branch 'bugfix_borrow_priority' into 'rel'

报文排队时同时计算primary profile和borrow profile的队列长度

See merge request tango/shaping-engine!71

5 files changed, 110 insertions, 10 deletions

diff --git a/shaping/include/shaper_stat.h b/shaping/include/shaper_stat.h
index a2527cd..d9ca979 100644
--- a/shaping/include/shaper_stat.h
+++ b/shaping/include/shaper_stat.h
@@ -50,6 +50,8 @@ struct shaping_stat* shaper_stat_init(int thread_num);
 
 void shaper_stat_queueing_pkt_inc(struct shaping_stat_for_profile *profile_stat, unsigned char direction, int thread_id);
 void shaper_stat_queueing_pkt_dec(struct shaping_stat_for_profile *profile_stat, unsigned char direction, int thread_id);
+void shaper_stat_queueing_pkt_inc_for_rule(struct shaping_rule_info *rule, unsigned char direction, int thread_id);
+void shaper_stat_queueing_pkt_dec_for_rule(struct shaping_rule_info *rule, unsigned char direction, int thread_id);
 void shaper_stat_forward_inc(struct shaping_stat_for_profile *profile_stat, unsigned char direction, int pkt_len, int thread_id);
 void shaper_stat_forward_all_rule_inc(struct shaping_stat *stat, struct shaping_flow *sf, unsigned char direction, int pkt_len, int thread_id);
 void shaper_stat_drop_inc(struct shaping_stat_for_profile *profile_stat, unsigned char direction, int thread_id);
diff --git a/shaping/src/shaper.cpp b/shaping/src/shaper.cpp
index 6ea2afe..9f99ffd 100644
--- a/shaping/src/shaper.cpp
+++ b/shaping/src/shaper.cpp
@@ -300,8 +300,7 @@ int shaper_flow_push(struct shaping_thread_ctx *ctx, struct shaping_flow *sf, un
         avl_tree_node_key_set(s_node->avl_node[priority], pkt_wrapper->income_time_ns);
         if (0 == avl_tree_node_insert(sp->priority_trees[priority], s_node->avl_node[priority])) {
             ret = 0;
-            //TODO: calculate queue_len for borrow profile and add judge when refresh stat????
-            //shaper_stat_queueing_pkt_inc(&s_rule_info->borrowing[i].stat, pkt_wrapper->direction, ctx->thread_index);
+            shaper_stat_queueing_pkt_inc(&s_rule_info->borrowing[i].stat, pkt_wrapper->direction, ctx->thread_index);
         }
     }
 
@@ -352,7 +351,7 @@ void shaper_flow_pop(struct shaping_thread_ctx *ctx, struct shaping_flow *sf)
         priority = s_rule_info->borrowing[i].priority;
         if (avl_node_in_tree(s_node->avl_node[priority])) {
             avl_tree_node_remove(sp->priority_trees[priority], s_node->avl_node[priority]);
-            //TODO: calculate queue_len for borrow profile and add judge when refresh stat????
+            shaper_stat_queueing_pkt_dec(&s_rule_info->borrowing[i].stat, pkt_wrapper->direction, ctx->thread_index);
         }
     }
 
@@ -993,7 +992,7 @@ static int shaper_polling_first_pkt_token_get(struct shaper *sp, struct shaping_
         return 0;
     } else {
         pkt_wrapper = shaper_first_pkt_get(sf);
-        shaper_stat_queueing_pkt_dec(&sf->matched_rule_infos[pkt_wrapper->rule_anchor].primary.stat, pkt_wrapper->direction, ctx->thread_index);
+        shaper_stat_queueing_pkt_dec_for_rule(&sf->matched_rule_infos[pkt_wrapper->rule_anchor], pkt_wrapper->direction, ctx->thread_index);
 
         sf->anchor = 0;
         if (shaper_flow_push(ctx, sf, pkt_wrapper->enqueue_time_us) != 0) {
@@ -1041,7 +1040,7 @@ void shaping_packet_process(struct shaping_thread_ctx *ctx, marsio_buff_t *rx_bu
     if (!shaper_queue_empty(sf)) {//already have queueing pkt, enqueue directly
         s_rule = &sf->matched_rule_infos[sf->anchor];
         if (0 == shaper_packet_enqueue(ctx, sf, rx_buff, meta)) {
-            shaper_stat_queueing_pkt_inc(&s_rule->primary.stat, meta->dir, ctx->thread_index);
+            shaper_stat_queueing_pkt_inc_for_rule(s_rule, meta->dir, ctx->thread_index);
             shaper_global_stat_queueing_inc(&ctx->thread_global_stat, meta->raw_len);
         } else {
             shaper_stat_drop_inc(&s_rule->primary.stat, meta->dir, ctx->thread_index);
diff --git a/shaping/src/shaper_stat.cpp b/shaping/src/shaper_stat.cpp
index 24edc08..656d2af 100644
--- a/shaping/src/shaper_stat.cpp
+++ b/shaping/src/shaper_stat.cpp
@@ -182,8 +182,10 @@ static void shaper_stat_profile_metirc_refresh(struct shaping_thread_ctx *ctx, i
     }
 
     if (need_update_guage) {
-        fieldstat_dynamic_table_metric_value_incrby(stat->instance, stat->table_id, stat->column_ids[IN_QUEUE_LEN_IDX], SHAPER_STAT_ROW_NAME, profile_stat->in.queue_len, tags, TAG_IDX_MAX, thread_id);
-        fieldstat_dynamic_table_metric_value_incrby(stat->instance, stat->table_id, stat->column_ids[OUT_QUEUE_LEN_IDX], SHAPER_STAT_ROW_NAME, profile_stat->out.queue_len, tags, TAG_IDX_MAX, thread_id);
+        if (profile_type == PROFILE_IN_RULE_TYPE_PRIMARY) {
+            fieldstat_dynamic_table_metric_value_incrby(stat->instance, stat->table_id, stat->column_ids[IN_QUEUE_LEN_IDX], SHAPER_STAT_ROW_NAME, profile_stat->in.queue_len, tags, TAG_IDX_MAX, thread_id);
+            fieldstat_dynamic_table_metric_value_incrby(stat->instance, stat->table_id, stat->column_ids[OUT_QUEUE_LEN_IDX], SHAPER_STAT_ROW_NAME, profile_stat->out.queue_len, tags, TAG_IDX_MAX, thread_id);
+        }
 
         struct shaping_hincrby_cb_arg *arg = (struct shaping_hincrby_cb_arg *)calloc(1, sizeof(struct shaping_hincrby_cb_arg));
         struct timespec curr_time;
@@ -221,7 +223,7 @@ void shaper_stat_refresh(struct shaping_thread_ctx *ctx, struct shaping_flow *sf
         need_refresh = 1;
     } else {
         clock_gettime(CLOCK_MONOTONIC, &curr_time);
-        if (curr_time.tv_sec - sf->stat_update_time.tv_sec > 0 || curr_time.tv_nsec - sf->stat_update_time.tv_nsec > SHAPER_STAT_REFRESH_TIME_NS) {
+        if (curr_time.tv_sec - sf->stat_update_time.tv_sec > 0 || curr_time.tv_nsec - sf->stat_update_time.tv_nsec >= SHAPER_STAT_REFRESH_TIME_NS) {
             need_refresh = 1;
             memcpy(&sf->stat_update_time, &curr_time, sizeof(struct timespec));
         }
@@ -308,6 +310,26 @@ void shaper_stat_queueing_pkt_dec(struct shaping_stat_for_profile *profile_stat,
     return;
 }
 
+void shaper_stat_queueing_pkt_inc_for_rule(struct shaping_rule_info *rule, unsigned char direction, int thread_id)
+{
+    shaper_stat_queueing_pkt_inc(&rule->primary.stat, direction, thread_id);
+    for (int i = 0; i < rule->borrowing_num; i++) {
+        shaper_stat_queueing_pkt_inc(&rule->borrowing[i].stat, direction, thread_id);
+    }
+
+    return;
+}
+
+void shaper_stat_queueing_pkt_dec_for_rule(struct shaping_rule_info *rule, unsigned char direction, int thread_id)
+{
+    shaper_stat_queueing_pkt_dec(&rule->primary.stat, direction, thread_id);
+    for (int i = 0; i < rule->borrowing_num; i++) {
+        shaper_stat_queueing_pkt_dec(&rule->borrowing[i].stat, direction, thread_id);
+    }
+
+    return;
+}
+
 void shaper_stat_max_latency_update(struct shaping_stat_for_profile *profile_stat, unsigned char direction, unsigned long long latency, int thread_id)
 {
     if (direction == SHAPING_DIR_IN) {
diff --git a/shaping/test/gtest_shaper.cpp b/shaping/test/gtest_shaper.cpp
index d72a410..4b0cda6 100644
--- a/shaping/test/gtest_shaper.cpp
+++ b/shaping/test/gtest_shaper.cpp
@@ -1591,7 +1591,7 @@ TEST(two_sessions, priority_non_block)
 profile_a(id 0): limit 1000
 profile_b(id 1): limit 1000
 */
-TEST(two_sessions, priority_block_borrow)
+TEST(two_sessions, borrow_when_primary_profile_priority_blocked)
 {
     struct stub_pkt_queue expec_tx_queue1;
     struct stub_pkt_queue expec_tx_queue2;
@@ -1653,6 +1653,82 @@ TEST(two_sessions, priority_block_borrow)
     stub_clear_matched_shaping_rules();
 }
 
+/*session1 match rule1; session2 match rule2
+  rule1:
+      priority:1
+      primary profile_a: (priority 1)
+      borrow  profile_b: (priority 2)
+  rule2:
+      priority:5
+      primary profile_b: (priority 5)
+
+profile_a(id 0): limit 0
+profile_b(id 1): limit 1000
+*/
+TEST(two_sessions, primary_profile_priority_blocked_by_borrow_profile)
+{
+    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[] = {2, 1};
+    int prioritys[] = {1, 5};
+    int profile_id[][MAX_REF_PROFILE] = {{0, 1}, {1}};
+    
+
+    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, 0, SHAPING_DIR_OUT);
+    stub_set_token_bucket_avl_per_sec(1, 1000, SHAPING_DIR_OUT);
+    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_OUT, &expec_tx_queue2, 1, 0);
+
+    ASSERT_EQ(0, judge_packet_eq(&expec_tx_queue1, actual_tx_queue, 10));
+    while (!TAILQ_EMPTY(&expec_tx_queue1)) {
+        stub_refresh_token_bucket(1);
+        polling_entry(ctx->thread_ctx[0].sp, ctx->thread_ctx[0].stat, &ctx->thread_ctx[0]);
+        polling_entry(ctx->thread_ctx[1].sp, ctx->thread_ctx[1].stat, &ctx->thread_ctx[1]);//blocked by priority, sf1 has priority 2 for profile_b(id 1)
+        stub_curr_time_ns_inc(STUB_TIME_INC_FOR_HMGET);
+
+        ASSERT_EQ(0, judge_packet_eq(&expec_tx_queue1, actual_tx_queue, 1));
+    }
+
+    while (!TAILQ_EMPTY(&expec_tx_queue2)) {
+        stub_refresh_token_bucket(1);
+        polling_entry(ctx->thread_ctx[1].sp, ctx->thread_ctx[1].stat, &ctx->thread_ctx[1]);
+
+        ASSERT_EQ(0, judge_packet_eq(&expec_tx_queue2, actual_tx_queue, 1));
+    }
+
+    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
   rule1:
       profile_a
@@ -1877,6 +1953,6 @@ TEST(statistics, udp_queueing_pkt)
 int main(int argc, char **argv) 
 {
     testing::InitGoogleTest(&argc, argv);
-    //testing::GTEST_FLAG(filter) = "single_session.dup_primary_profile";
+    //testing::GTEST_FLAG(filter) = "two_sessions.primary_profile_priority_blocked_by_borrow_profile";
     return RUN_ALL_TESTS();
 }
\ No newline at end of file
diff --git a/shaping/test/stub.h b/shaping/test/stub.h
index f7119d2..e4c0476 100644
--- a/shaping/test/stub.h
+++ b/shaping/test/stub.h
@@ -10,6 +10,7 @@
 #define STUB_MAAT_SHAPING_PROFILE_TABLE_ID 1
 
 #define STUB_TIME_INC_FOR_PACKET 1000000
+#define STUB_TIME_INC_FOR_HMGET 10000000
 
 #define STUB_TEST_VSYS_ID 2333