1.异步转发增加限制不让本地token负值太高

2.修复generic profile请求token放大倍数不生效

2 files changed, 67 insertions, 62 deletions

diff --git a/shaping/include/shaper.h b/shaping/include/shaper.h
index 62f0eb6..ef27c2c 100644
--- a/shaping/include/shaper.h
+++ b/shaping/include/shaper.h
@@ -107,7 +107,7 @@ enum shaping_profile_limit_direction {
 
 struct shaper_token_multiple {
     int token_get_multiple;
-    unsigned char has_drop_by_queue_full;
+    unsigned char token_not_enough;
     unsigned char has_failed_get_token;
     time_t token_multiple_update_time_s;
 };
@@ -119,7 +119,7 @@ struct shaping_profile_hash_node {
     long long in_deposit_token_bits[SHAPING_PRIORITY_NUM_MAX];
     long long out_deposit_token_bits[SHAPING_PRIORITY_NUM_MAX];
     long long bidirection_deposit_token_bits[SHAPING_PRIORITY_NUM_MAX];
-    long long last_failed_get_token_ms;
+    long long last_failed_get_token_ms[SHAPING_DIR_MAX];
     long long last_hmget_ms;
     long long queue_len[SHAPING_PRIORITY_NUM_MAX];
     long long local_queue_len[SHAPING_PRIORITY_NUM_MAX];
@@ -132,7 +132,7 @@ struct shaping_profile_hash_node {
     struct shaper_aqm_blue_para aqm_blue_para;
     struct shaper_aqm_codel_para aqm_codel_para;
     unsigned char is_invalid;
-    unsigned char async_pass[SHAPING_PRIORITY_NUM_MAX][SHAPING_DIR_MAX];
+    unsigned char async_pass[SHAPING_DIR_MAX];
     struct timeout timeout_handle;
     UT_hash_handle hh;
 };
@@ -145,7 +145,7 @@ struct shaping_profile_info {
     long long in_deposit_token_bits;
     long long out_deposit_token_bits;
     long long bidirection_deposit_token_bits;
-    long long last_failed_get_token_ms;
+    long long last_failed_get_token_ms[SHAPING_DIR_MAX];
     unsigned long long enqueue_time_us;//to calculate max latency
     struct shaping_stat_for_profile stat;
     struct shaping_profile_hash_node *hash_node;
diff --git a/shaping/src/shaper.cpp b/shaping/src/shaper.cpp
index 16c9475..b51e717 100644
--- a/shaping/src/shaper.cpp
+++ b/shaping/src/shaper.cpp
@@ -379,17 +379,16 @@ int shaper_flow_in_order_get(struct shaper *sp, struct shaper_flow_instance sf_i
         avl_node = avl_tree_next_in_order_node_get(avl_node);
     }
 
-
     return count;
 }
 
-static void shaper_profile_async_pass_set(struct shaping_profile_info *profile, unsigned char direction, int priority, int async_pass_enabled)
+static void shaper_profile_async_pass_set(struct shaping_profile_info *profile, unsigned char direction, int async_pass_enabled)
 {
     struct shaping_profile_hash_node *pf_hash_node = profile->hash_node;
     unsigned char *async_pass = NULL;
 
     if (profile->type == PROFILE_TYPE_GENERIC) {
-        async_pass = &pf_hash_node->async_pass[priority][direction];
+        async_pass = &pf_hash_node->async_pass[direction];
     } else {
         async_pass = &profile->async_pass[direction];
     }
@@ -401,18 +400,54 @@ static void shaper_profile_async_pass_set(struct shaping_profile_info *profile,
     return;
 }
 
-static int shaper_profile_async_pass_get(struct shaping_profile_info *profile, unsigned char direction, int priority)
+static int shaper_profile_async_pass_get(struct shaping_profile_info *profile, unsigned char direction)
 {
     struct shaping_profile_hash_node *pf_hash_node = profile->hash_node;
 
     if (profile->type == PROFILE_TYPE_GENERIC) {
-        return pf_hash_node->async_pass[priority][direction];
+        return pf_hash_node->async_pass[direction];
     } else {
         return profile->async_pass[direction];
     }
 }
 
-static void shaper_deposit_token_add(struct shaping_profile_info *profile, int req_token_bits, unsigned char direction, int priority)
+static void shaper_token_multiple_update(struct shaping_thread_ctx *ctx, struct shaping_profile_info *profile)
+{
+    if (profile->type != PROFILE_TYPE_GENERIC) {
+        return;
+    }
+
+    struct shaper_token_multiple *token_multiple = &profile->hash_node->token_multiple;
+    int curr_multiple = token_multiple->token_get_multiple;
+    time_t curr_time_s = time(NULL);
+    int token_multiple_min = ctx->conf.token_multiple_min;
+    int token_multiple_max = ctx->conf.token_multiple_max;
+
+    if (curr_time_s - token_multiple->token_multiple_update_time_s < TOKEN_MULTIPLE_UPDATE_INTERVAL_S) {
+        return;
+    }
+
+    token_multiple->token_multiple_update_time_s = curr_time_s;
+
+    if (token_multiple->has_failed_get_token) {
+        token_multiple->token_get_multiple = (curr_multiple - 1) < token_multiple_min ? token_multiple_min : (curr_multiple - 1);
+        goto END;
+    }
+
+    if (token_multiple->token_not_enough) {
+        token_multiple->token_get_multiple = (curr_multiple + 1) > token_multiple_max ? token_multiple_max : (curr_multiple + 1);
+        goto END;
+    }
+
+END:
+    LOG_INFO("%s: profile id %d, token_get_multiple %d, has_failed_get_token %d, token_not_enough %d", LOG_TAG_SHAPING, profile->id, token_multiple->token_get_multiple, token_multiple->has_failed_get_token, token_multiple->token_not_enough);
+    token_multiple->has_failed_get_token = 0;
+    token_multiple->token_not_enough = 0;
+    
+    return;
+}
+
+static void shaper_deposit_token_add(struct shaping_thread_ctx *ctx, struct shaping_profile_info *profile, int req_token_bits, unsigned char direction, int priority)
 {
     long long *deposit_token;
     struct shaping_profile_hash_node *pf_hash_node = profile->hash_node;
@@ -445,44 +480,11 @@ static void shaper_deposit_token_add(struct shaping_profile_info *profile, int r
 
     *deposit_token += req_token_bits;
     if (*deposit_token > 0) {
-        shaper_profile_async_pass_set(profile, direction, priority, 1);
-    }
-}
-
-static void shaper_token_multiple_update(struct shaping_thread_ctx *ctx, struct shaping_profile_info *profile)
-{
-    if (profile->type != PROFILE_TYPE_GENERIC) {
-        return;
-    }
-
-    struct shaper_token_multiple *token_multiple = &profile->hash_node->token_multiple;
-    int curr_multiple = token_multiple->token_get_multiple;
-    time_t curr_time_s = time(NULL);
-    int token_multiple_min = ctx->conf.token_multiple_min;
-    int token_multiple_max = ctx->conf.token_multiple_max;
-
-    if (curr_time_s - token_multiple->token_multiple_update_time_s < TOKEN_MULTIPLE_UPDATE_INTERVAL_S) {
-        return;
-    }
-
-    token_multiple->token_multiple_update_time_s = curr_time_s;
-
-    if (token_multiple->has_failed_get_token) {
-        token_multiple->token_get_multiple = (curr_multiple - 1) < token_multiple_min ? token_multiple_min : (curr_multiple - 1);
-        goto END;
-    }
-
-    if (token_multiple->has_drop_by_queue_full) {
-        token_multiple->token_get_multiple = (curr_multiple + 1) > token_multiple_max ? token_multiple_max : (curr_multiple + 1);
-        goto END;
+        shaper_profile_async_pass_set(profile, direction, 1);
+    } else {
+        pf_hash_node->token_multiple.token_not_enough = 1;
+        shaper_token_multiple_update(ctx, profile);
     }
-
-END:
-    LOG_INFO("%s: profile id %d, token_get_multiple %d, has_failed_get_token %d, has_drop_by_queue_full %d", LOG_TAG_SHAPING, profile->id, token_multiple->token_get_multiple, token_multiple->has_failed_get_token, token_multiple->has_drop_by_queue_full);
-    token_multiple->has_failed_get_token = 0;
-    token_multiple->has_drop_by_queue_full = 0;
-    
-    return;
 }
 
 static void shaper_token_get_cb(const struct swarmkv_reply *reply, void * cb_arg)
@@ -519,11 +521,11 @@ static void shaper_token_get_cb(const struct swarmkv_reply *reply, void * cb_arg
     }
 
     if (reply->integer > 0) {
-        shaper_deposit_token_add(profile, reply->integer, arg->direction, profile->priority);//deposit tokens to profile
+        shaper_deposit_token_add(ctx, profile, reply->integer, arg->direction, profile->priority);//deposit tokens to profile
     }
 
     if (reply->integer == 0) {
-        shaper_profile_async_pass_set(profile, arg->direction, profile->priority, 0);
+        shaper_profile_async_pass_set(profile, arg->direction, 0);
 
         if (profile->type == PROFILE_TYPE_GENERIC) {
             pf_hash_node->token_multiple.has_failed_get_token = 1;
@@ -537,12 +539,12 @@ END:
     if (reply->type != SWARMKV_REPLY_INTEGER || reply->integer == 0) {
         switch (profile->type) {
             case PROFILE_TYPE_GENERIC:
-                pf_hash_node->last_failed_get_token_ms = curr_time.tv_sec * MILLI_SECONDS_PER_SEC + curr_time.tv_nsec / NANO_SECONDS_PER_MILLI_SEC;
+                pf_hash_node->last_failed_get_token_ms[arg->direction] = curr_time.tv_sec * MILLI_SECONDS_PER_SEC + curr_time.tv_nsec / NANO_SECONDS_PER_MILLI_SEC;
                 break;
             case PROFILE_TYPE_HOST_FARINESS:
             case PROFILE_TYPE_MAX_MIN_HOST_FAIRNESS:
             case PROFILE_TYPE_SPLIT_BY_LOCAL_HOST:
-                profile->last_failed_get_token_ms = curr_time.tv_sec * MILLI_SECONDS_PER_SEC + curr_time.tv_nsec / NANO_SECONDS_PER_MILLI_SEC;
+                profile->last_failed_get_token_ms[arg->direction] = curr_time.tv_sec * MILLI_SECONDS_PER_SEC + curr_time.tv_nsec / NANO_SECONDS_PER_MILLI_SEC;
                 break;
         }
     }
@@ -560,6 +562,7 @@ static int shaper_deposit_token_get(struct shaping_profile_info *profile, int re
     long long *deposit_token;
     struct shaping_profile_hash_node *pf_hash_node = profile->hash_node;
     int ret = -1;
+    int token_multiple;
 
     switch (profile->type) {
         case PROFILE_TYPE_GENERIC:
@@ -570,6 +573,7 @@ static int shaper_deposit_token_get(struct shaping_profile_info *profile, int re
             } else {
                 deposit_token = &pf_hash_node->out_deposit_token_bits[priority];
             }
+            token_multiple = pf_hash_node->token_multiple.token_get_multiple;
             break;
         case PROFILE_TYPE_HOST_FARINESS:
         case PROFILE_TYPE_MAX_MIN_HOST_FAIRNESS:
@@ -581,6 +585,7 @@ static int shaper_deposit_token_get(struct shaping_profile_info *profile, int re
             } else {
                 deposit_token = &profile->out_deposit_token_bits;
             }
+            token_multiple = TOKEN_MULTIPLE_DEFAULT;
             break;
         default:
             LOG_ERROR("%s: invalid profile type %d, profile id %d", LOG_TAG_SHAPING, profile->type, profile->id);
@@ -596,6 +601,10 @@ static int shaper_deposit_token_get(struct shaping_profile_info *profile, int re
         ret = 0;
     }
 
+    if (*deposit_token + (req_token_bits * token_multiple * 2) < 0) {
+        shaper_profile_async_pass_set(profile, direction, 0);
+    }
+
     return ret;
 }
 
@@ -758,8 +767,8 @@ static int shaper_profile_is_priority_blocked(struct shaping_thread_ctx *ctx, st
 
 END:
     if (curr_time_ms - profile->hash_node->priority_blocked_time_ms[priority] < PRIORITY_BLOCK_MIN_TIME_MS) {
-        shaper_profile_async_pass_set(profile, SHAPING_DIR_OUT, priority, 0);
-        shaper_profile_async_pass_set(profile, SHAPING_DIR_IN, priority, 0);
+        shaper_profile_async_pass_set(profile, SHAPING_DIR_OUT, 0);
+        shaper_profile_async_pass_set(profile, SHAPING_DIR_IN, 0);
         return 1;
     } else {
         return 0;
@@ -786,18 +795,18 @@ void shaper_profile_hash_node_set(struct shaping_thread_ctx *ctx, struct shaping
     return;
 }
 
-static int shaping_swarmkv_is_too_short_interval(long long curr_time_ms, struct shaping_profile_info *profile)
+static int shaping_swarmkv_is_too_short_interval(long long curr_time_ms, struct shaping_profile_info *profile, unsigned char direction)
 {
     long long last_failed_ms = 0;
 
     switch (profile->type) {
         case PROFILE_TYPE_GENERIC:
-            last_failed_ms = profile->hash_node->last_failed_get_token_ms;
+            last_failed_ms = profile->hash_node->last_failed_get_token_ms[direction];
             break;
         case PROFILE_TYPE_HOST_FARINESS:
         case PROFILE_TYPE_MAX_MIN_HOST_FAIRNESS:
         case PROFILE_TYPE_SPLIT_BY_LOCAL_HOST:
-            last_failed_ms = profile->last_failed_get_token_ms;
+            last_failed_ms = profile->last_failed_get_token_ms[direction];
             break;
     }
 
@@ -829,7 +838,7 @@ static int shaper_token_consume(struct shaping_thread_ctx *ctx, struct shaping_f
         return SHAPER_TOKEN_GET_PASS;//rule is disabled, don't need to get token and forward packet
     }
 
-    if (shaper_profile_async_pass_get(profile, direction, profile->priority) == 1) {
+    if (shaper_profile_async_pass_get(profile, direction) == 1) {
         shaper_deposit_token_get(profile, req_token_bytes * 8, direction, profile->priority, 1, &need_get_token);
         ret = SHAPER_TOKEN_GET_SUCCESS;
     } else if (shaper_deposit_token_get(profile, req_token_bytes * 8, direction, profile->priority, 0, &need_get_token) == 0) {
@@ -842,7 +851,7 @@ static int shaper_token_consume(struct shaping_thread_ctx *ctx, struct shaping_f
 
     long long curr_time_ms = curr_timespec->tv_sec * MILLI_SECONDS_PER_SEC + curr_timespec->tv_nsec / NANO_SECONDS_PER_MILLI_SEC;
 
-    if (shaping_swarmkv_is_too_short_interval(curr_time_ms, profile)) {
+    if (shaping_swarmkv_is_too_short_interval(curr_time_ms, profile, direction)) {
         return ret;
     }
 
@@ -1036,8 +1045,6 @@ DROP:
     shaper_stat_drop_inc(&pf_info->stat, meta->dir, ctx->thread_index);
     sf->anchor = 0;
 
-    pf_info->hash_node->token_multiple.has_drop_by_queue_full = 1;
-    shaper_token_multiple_update(ctx, pf_info);
     return SHAPING_DROP;
 }
 
@@ -1156,8 +1163,6 @@ void shaping_packet_process(struct shaping_thread_ctx *ctx, marsio_buff_t *rx_bu
             shaper_global_stat_queueing_inc(&ctx->thread_global_stat, meta->raw_len);
         } else {
             struct shaping_profile_info *pf_info = &s_rule->primary;
-            pf_info->hash_node->token_multiple.has_drop_by_queue_full = 1;
-            shaper_token_multiple_update(ctx, pf_info);
 
             shaper_stat_drop_inc(&pf_info->stat, meta->dir, ctx->thread_index);
             shaper_global_stat_drop_inc(&ctx->thread_global_stat, meta->raw_len);