From 8f9592d1c0066738f97932213dfcdf32271c34c2 Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Wed, 20 Oct 2021 19:05:17 +0800
Subject: [PATCH 01/14] finish coding

---
 ...oken_bucket_throttling_controller_test.cpp | 169 ++++++++++++++++
 .../token_bucket_throttling_controller.cpp    | 182 ++++++++++++++++++
 2 files changed, 351 insertions(+)
 create mode 100644 src/utils/test/token_bucket_throttling_controller_test.cpp
 create mode 100644 src/utils/token_bucket_throttling_controller.cpp

diff --git a/src/utils/test/token_bucket_throttling_controller_test.cpp b/src/utils/test/token_bucket_throttling_controller_test.cpp
new file mode 100644
index 0000000000..929fa5decc
--- /dev/null
+++ b/src/utils/test/token_bucket_throttling_controller_test.cpp
@@ -0,0 +1,169 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <dsn/utils/token_bucket_throttling_controller.h>
+#include <dsn/utility/TokenBucket.h>
+
+#include <gtest/gtest.h>
+
+namespace dsn {
+namespace utils {
+
+class token_bucket_throttling_controller_test : public ::testing::Test
+{
+public:
+    void test_parse_env_basic_token_bucket_throttling()
+    {
+        token_bucket_throttling_controller cntl;
+        std::string parse_err;
+        bool env_changed = false;
+        std::string old_value;
+
+        int partition_count = 4;
+        std::string old_env = "";
+        std::string env = "20000*delay*100";
+        // token_bucket_throttling_controller doesn't support delay only
+        ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
+        ASSERT_EQ(cntl._env_value, env);
+        ASSERT_EQ(cntl._partition_count, 4);
+        ASSERT_EQ(cntl._burstsize, 0);
+        ASSERT_EQ(cntl._rate, 0);
+        ASSERT_EQ(cntl._enabled, false);
+        ASSERT_EQ(env_changed, true);
+        ASSERT_EQ(old_value, old_env);
+        ASSERT_EQ(parse_err, "");
+
+        old_env = env;
+        env = "200K";
+        ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
+        ASSERT_EQ(cntl._enabled, true);
+        ASSERT_EQ(cntl._env_value, env);
+        ASSERT_EQ(cntl._partition_count, 4);
+        ASSERT_EQ(cntl._rate, 200000 / partition_count);
+        ASSERT_EQ(cntl._burstsize, 200000 / partition_count);
+        ASSERT_EQ(env_changed, true);
+        ASSERT_EQ(old_value, old_env);
+        ASSERT_EQ(parse_err, "");
+
+        old_env = env;
+        env = "20000*delay*100,20000*reject*100";
+        ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
+        ASSERT_EQ(cntl._enabled, true);
+        ASSERT_EQ(cntl._env_value, env);
+        ASSERT_EQ(cntl._partition_count, 4);
+        ASSERT_EQ(cntl._rate, 20000 / partition_count);
+        ASSERT_EQ(cntl._burstsize, 20000 / partition_count);
+        ASSERT_EQ(env_changed, true);
+        ASSERT_EQ(old_value, old_env);
+        ASSERT_EQ(parse_err, "");
+
+        old_env = env;
+        env = "20000*reject*100";
+        ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
+        ASSERT_EQ(cntl._enabled, true);
+        ASSERT_EQ(cntl._env_value, env);
+        ASSERT_EQ(cntl._partition_count, 4);
+        ASSERT_EQ(cntl._rate, 20000 / partition_count);
+        ASSERT_EQ(cntl._burstsize, 20000 / partition_count);
+        ASSERT_EQ(env_changed, true);
+        ASSERT_EQ(old_value, old_env);
+        ASSERT_EQ(parse_err, "");
+
+        // invalid argument
+        old_env = env;
+        env = "*deldday*100";
+        ASSERT_FALSE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
+        ASSERT_EQ(env_changed, false);
+        ASSERT_EQ(parse_err, "wrong format, you can set like 20000 or 20K");
+        ASSERT_EQ(cntl._enabled, true); // ensure invalid env won't stop throttling
+        ASSERT_EQ(old_value, old_env);
+
+        ASSERT_FALSE(cntl.parse_from_env("", 4, parse_err, env_changed, old_value));
+        ASSERT_EQ(env_changed, false);
+        ASSERT_NE(parse_err, "");
+        ASSERT_EQ(parse_err, "wrong format, you can set like 20000 or 20K");
+        ASSERT_EQ(cntl._enabled, true);
+        ASSERT_EQ(old_value, old_env);
+    }
+
+    void throttle_test()
+    {
+        auto cntl = std::make_unique<token_bucket_throttling_controller>();
+        std::string parse_err;
+        bool env_changed = false;
+        std::string old_value;
+        const int partition_count = 4;
+
+        int throttle_limit = 200000;
+        cntl->parse_from_env(
+            std::to_string(throttle_limit), partition_count, parse_err, env_changed, old_value);
+
+        auto token_bucket = std::make_unique<DynamicTokenBucket>();
+        int fail_count = 0;
+        for (int i = 0; i < 100000; i++) {
+            token_bucket->consumeWithBorrowAndWait(
+                1, throttle_limit / partition_count * 0.8, throttle_limit / partition_count * 1.0);
+            if (!cntl->control(1)) {
+                fail_count++;
+            }
+        }
+        ASSERT_EQ(fail_count, 0);
+
+        sleep(1);
+
+        fail_count = 0;
+        for (int i = 0; i < 100000; i++) {
+            token_bucket->consumeWithBorrowAndWait(
+                1, throttle_limit / partition_count * 1.2, throttle_limit / partition_count * 1.5);
+            if (!cntl->control(1)) {
+                fail_count++;
+            }
+        }
+        ASSERT_GT(fail_count, 10000);
+
+        sleep(1);
+
+        fail_count = 0;
+        int fail_count1 = 0;
+        for (int i = 0; i < 200000; i++) {
+            if (i < 100000) {
+                token_bucket->consumeWithBorrowAndWait(1,
+                                                       throttle_limit / partition_count * 1.2,
+                                                       throttle_limit / partition_count * 1.5);
+                fail_count1 = fail_count;
+            } else {
+                token_bucket->consumeWithBorrowAndWait(1,
+                                                       throttle_limit / partition_count * 0.2,
+                                                       throttle_limit / partition_count * 0.3);
+            }
+            if (!cntl->control(1)) {
+                fail_count++;
+            }
+        }
+        ASSERT_GT(fail_count1, 10000);
+        ASSERT_LE(fail_count, fail_count1 * 1.2);
+    }
+};
+
+TEST_F(token_bucket_throttling_controller_test, test_parse_env_basic_token_bucket_throttling)
+{
+    test_parse_env_basic_token_bucket_throttling();
+}
+
+TEST_F(token_bucket_throttling_controller_test, throttle_test) { throttle_test(); }
+} // namespace utils
+} // namespace dsn
diff --git a/src/utils/token_bucket_throttling_controller.cpp b/src/utils/token_bucket_throttling_controller.cpp
new file mode 100644
index 0000000000..34e61cfe42
--- /dev/null
+++ b/src/utils/token_bucket_throttling_controller.cpp
@@ -0,0 +1,182 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <dsn/utils/token_bucket_throttling_controller.h>
+
+namespace dsn {
+namespace utils {
+
+token_bucket_throttling_controller::token_bucket_throttling_controller()
+    : _enabled(false), _partition_count(0), _rate(0), _burstsize(0)
+{
+    _token_bucket = std::make_unique<DynamicTokenBucket>();
+    is_get_perfcounter = false;
+}
+
+token_bucket_throttling_controller::token_bucket_throttling_controller(
+    dsn::perf_counter_wrapper *reject_task_counter)
+    : _enabled(false), _partition_count(0), _rate(0), _burstsize(0)
+{
+    _token_bucket = std::make_unique<DynamicTokenBucket>();
+
+    dassert(reject_task_counter, "reject_task_counter == nullptr");
+    _reject_task_counter = reject_task_counter;
+    is_get_perfcounter = true;
+}
+
+void token_bucket_throttling_controller::only_count(int32_t request_units)
+{
+    if (!_enabled) {
+        return;
+    }
+    _token_bucket->consumeWithBorrowNonBlocking((double)request_units, _rate, _burstsize);
+}
+
+bool token_bucket_throttling_controller::control(int32_t request_units = 1)
+{
+    if (!_enabled) {
+        return true;
+    }
+    auto res =
+        _token_bucket->consumeWithBorrowNonBlocking((double)request_units, _rate, _burstsize);
+
+    if (res.get_value_or(0) > 0) {
+        if (is_get_perfcounter) {
+            _reject_task_counter->operator->()->increment();
+        }
+        return false;
+    }
+    return true;
+}
+
+void token_bucket_throttling_controller::reset(bool &changed, std::string &old_env_value)
+{
+    if (_enabled) {
+        changed = true;
+        old_env_value = _env_value;
+        _enabled = false;
+        _env_value.clear();
+        _partition_count = 0;
+        _rate = 0;
+        _burstsize = 0;
+    } else {
+        changed = false;
+    }
+}
+
+// return the current env value.
+const std::string &token_bucket_throttling_controller::env_value() const { return _env_value; }
+
+bool token_bucket_throttling_controller::parse_from_env(const std::string &env_value,
+                                                        int partition_count,
+                                                        std::string &parse_error,
+                                                        bool &changed,
+                                                        std::string &old_env_value)
+{
+    old_env_value = _env_value;
+    changed = false;
+
+    if (_enabled && env_value == _env_value && partition_count == _partition_count)
+        return true;
+
+    int64_t reject_size_value;
+    bool enabled;
+    if (!transform_env_string(env_value, reject_size_value, enabled, parse_error)) {
+        return false;
+    }
+
+    changed = true;
+
+    _enabled = enabled;
+    _env_value = env_value;
+    _partition_count = partition_count;
+    _rate = reject_size_value / (partition_count > 0 ? partition_count : 1);
+    _burstsize = _rate;
+    return true;
+}
+
+bool token_bucket_throttling_controller::string_to_value(std::string str, int64_t &value)
+{
+    int64_t unit_multiplier = 1;
+    if (*str.rbegin() == 'M') {
+        unit_multiplier = 1000 * 1000;
+    } else if (*str.rbegin() == 'K') {
+        unit_multiplier = 1000;
+    }
+    if (unit_multiplier != 1) {
+        str.pop_back();
+    }
+    if (!buf2int64(str, value) || value < 0) {
+        return false;
+    }
+    value *= unit_multiplier;
+    return true;
+}
+
+bool token_bucket_throttling_controller::validate(const std::string &env, std::string &hint_message)
+{
+    int64_t temp;
+    bool temp_bool;
+    bool validated = transform_env_string(env, temp, temp_bool, hint_message);
+    return validated;
+};
+
+bool token_bucket_throttling_controller::transform_env_string(const std::string &env,
+                                                              int64_t &reject_size_value,
+                                                              bool &enabled,
+                                                              std::string &hint_message)
+{
+    enabled = true;
+
+    if (buf2int64(env, reject_size_value) && reject_size_value >= 0) {
+        return true;
+    }
+
+    // format like "200K"
+    if (string_to_value(env, reject_size_value)) {
+        return true;
+    }
+
+    // format like "20000*delay*100"
+    if (env.find("delay") != -1 && env.find("reject") == -1) {
+        reject_size_value = 0;
+        enabled = false;
+
+        dinfo("token_bucket_throttling_controller doesn't support delay method, so throttling "
+              "controller is disabled now");
+        return true;
+    }
+
+    // format like "20000*delay*100,20000*reject*100"
+    auto comma_index = env.find(",");
+    auto star_index = env.find("*reject", comma_index + 1);
+    if (star_index < 0) {
+        hint_message = "wrong format, you can set like 20000 or 20K";
+        return false;
+    }
+    auto reject_size = env.substr(comma_index + 1, star_index - comma_index - 1);
+
+    if (string_to_value(reject_size, reject_size_value)) {
+        return true;
+    }
+
+    hint_message = "wrong format, you can set like 20000 or 20K";
+    return false;
+}
+
+} // namespace utils
+} // namespace dsn

From 6a9a0800fb952c7ea026c0f7b623f9882906c420 Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Wed, 20 Oct 2021 19:38:40 +0800
Subject: [PATCH 02/14] add some comments and change the interface

---
 .../token_bucket_throttling_controller.h      | 84 +++++++++++++++++++
 .../token_bucket_throttling_controller.cpp    | 15 ----
 2 files changed, 84 insertions(+), 15 deletions(-)
 create mode 100644 include/dsn/utils/token_bucket_throttling_controller.h

diff --git a/include/dsn/utils/token_bucket_throttling_controller.h b/include/dsn/utils/token_bucket_throttling_controller.h
new file mode 100644
index 0000000000..a93b6c4c33
--- /dev/null
+++ b/include/dsn/utils/token_bucket_throttling_controller.h
@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <stdint.h>
+#include <string>
+#include <dsn/utility/TokenBucket.h>
+#include <dsn/dist/fmt_logging.h>
+#include <dsn/utility/strings.h>
+#include <dsn/utility/string_conv.h>
+#include <dsn/c/api_layer1.h>
+#include <dsn/perf_counter/perf_counter_wrapper.h>
+
+namespace dsn {
+namespace utils {
+
+using DynamicTokenBucket = folly::BasicDynamicTokenBucket<std::chrono::steady_clock>;
+
+// token_bucket_throttling_controller ignores `delay` parameter
+class token_bucket_throttling_controller
+{
+private:
+    friend class token_bucket_throttling_controller_test;
+
+    std::unique_ptr<DynamicTokenBucket> _token_bucket;
+
+    bool _enabled;
+    std::string _env_value;
+    int32_t _partition_count = 0;
+    double _rate;
+    double _burstsize;
+
+    dsn::perf_counter_wrapper *_reject_task_counter;
+
+public:
+    token_bucket_throttling_controller();
+
+    // Non-blocking, if exceed limits, return false.
+    bool control(int32_t request_units);
+
+    // Non-blocking, only count for the tokenbucket, never return err
+    void only_count(int32_t request_units);
+
+    void reset(bool &changed, std::string &old_env_value);
+
+    // return the current env value.
+    const std::string &env_value() const;
+
+    bool parse_from_env(const std::string &env_value,
+                        int partition_count,
+                        std::string &parse_error,
+                        bool &changed,
+                        std::string &old_env_value);
+
+    static bool string_to_value(std::string str, int64_t &value);
+
+    // support format:
+    // env == 20000*delay*100,20000*reject*100 (historical format)
+    // env == 20000/20K/2M (new format)
+    static bool validate(const std::string &env, std::string &hint_message);
+
+    static bool transform_env_string(const std::string &env,
+                                     int64_t &reject_size_value,
+                                     bool &enabled,
+                                     std::string &hint_message);
+};
+
+} // namespace utils
+} // namespace dsn
diff --git a/src/utils/token_bucket_throttling_controller.cpp b/src/utils/token_bucket_throttling_controller.cpp
index 34e61cfe42..0d0ed2debf 100644
--- a/src/utils/token_bucket_throttling_controller.cpp
+++ b/src/utils/token_bucket_throttling_controller.cpp
@@ -24,18 +24,6 @@ token_bucket_throttling_controller::token_bucket_throttling_controller()
     : _enabled(false), _partition_count(0), _rate(0), _burstsize(0)
 {
     _token_bucket = std::make_unique<DynamicTokenBucket>();
-    is_get_perfcounter = false;
-}
-
-token_bucket_throttling_controller::token_bucket_throttling_controller(
-    dsn::perf_counter_wrapper *reject_task_counter)
-    : _enabled(false), _partition_count(0), _rate(0), _burstsize(0)
-{
-    _token_bucket = std::make_unique<DynamicTokenBucket>();
-
-    dassert(reject_task_counter, "reject_task_counter == nullptr");
-    _reject_task_counter = reject_task_counter;
-    is_get_perfcounter = true;
 }
 
 void token_bucket_throttling_controller::only_count(int32_t request_units)
@@ -55,9 +43,6 @@ bool token_bucket_throttling_controller::control(int32_t request_units = 1)
         _token_bucket->consumeWithBorrowNonBlocking((double)request_units, _rate, _burstsize);
 
     if (res.get_value_or(0) > 0) {
-        if (is_get_perfcounter) {
-            _reject_task_counter->operator->()->increment();
-        }
         return false;
     }
     return true;

From cc61379d594b91123d2c8bb933008c55381bd1ca Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Fri, 22 Oct 2021 17:41:56 +0800
Subject: [PATCH 03/14] add comments & change codes

---
 .../token_bucket_throttling_controller.h      | 43 +++++++++++--------
 ...oken_bucket_throttling_controller_test.cpp | 18 ++++++--
 .../token_bucket_throttling_controller.cpp    | 20 +++------
 3 files changed, 48 insertions(+), 33 deletions(-)

diff --git a/include/dsn/utils/token_bucket_throttling_controller.h b/include/dsn/utils/token_bucket_throttling_controller.h
index a93b6c4c33..e0b1060a1c 100644
--- a/include/dsn/utils/token_bucket_throttling_controller.h
+++ b/include/dsn/utils/token_bucket_throttling_controller.h
@@ -17,14 +17,7 @@
 
 #pragma once
 
-#include <stdint.h>
-#include <string>
 #include <dsn/utility/TokenBucket.h>
-#include <dsn/dist/fmt_logging.h>
-#include <dsn/utility/strings.h>
-#include <dsn/utility/string_conv.h>
-#include <dsn/c/api_layer1.h>
-#include <dsn/perf_counter/perf_counter_wrapper.h>
 
 namespace dsn {
 namespace utils {
@@ -45,22 +38,40 @@ class token_bucket_throttling_controller
     double _rate;
     double _burstsize;
 
-    dsn::perf_counter_wrapper *_reject_task_counter;
-
 public:
     token_bucket_throttling_controller();
 
-    // Non-blocking, if exceed limits, return false.
-    bool control(int32_t request_units);
-
-    // Non-blocking, only count for the tokenbucket, never return err
-    void only_count(int32_t request_units);
+    // return ture means you can get token
+    // return false means the bucket is already empty, but the token is borrowed from future.
+    // non-blocking
+    bool get_token(int32_t request_units);
 
+    // reset to no throttling.
     void reset(bool &changed, std::string &old_env_value);
 
     // return the current env value.
     const std::string &env_value() const;
 
+    // Configures throttling strategy dynamically from app-envs.
+    //
+    // Support two style format:
+    // 1. style: "20000*delay*100,20000*reject*100"
+    //      example: 20000*delay*100,20000*reject*100
+    //      result: reject 20000 request_units, but never delay
+    //      example: 20000*delay*100
+    //      result: never result or delay
+    //      example: 20000*reject*100
+    //      result: reject 20000 request_units
+    // 2. style: 20/"20K"/"20M"
+    //      example: 20K
+    //      result reject 20000 request_units
+    //
+    // return true if parse succeed.
+    // return false if parse failed for the reason of invalid env_value.
+    // if return false, the original value will not be changed.
+    // 'parse_error' is set when return false.
+    // 'changed' is set when return true.
+    // 'old_env_value' is set when 'changed' is set to true.
     bool parse_from_env(const std::string &env_value,
                         int partition_count,
                         std::string &parse_error,
@@ -69,9 +80,7 @@ class token_bucket_throttling_controller
 
     static bool string_to_value(std::string str, int64_t &value);
 
-    // support format:
-    // env == 20000*delay*100,20000*reject*100 (historical format)
-    // env == 20000/20K/2M (new format)
+    // wrapper of transform_env_string, check if the env string is validated.
     static bool validate(const std::string &env, std::string &hint_message);
 
     static bool transform_env_string(const std::string &env,
diff --git a/src/utils/test/token_bucket_throttling_controller_test.cpp b/src/utils/test/token_bucket_throttling_controller_test.cpp
index 929fa5decc..07d325300d 100644
--- a/src/utils/test/token_bucket_throttling_controller_test.cpp
+++ b/src/utils/test/token_bucket_throttling_controller_test.cpp
@@ -71,6 +71,18 @@ class token_bucket_throttling_controller_test : public ::testing::Test
         ASSERT_EQ(old_value, old_env);
         ASSERT_EQ(parse_err, "");
 
+        old_env = env;
+        env = "20K*delay*100,20K*reject*100";
+        ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
+        ASSERT_EQ(cntl._enabled, true);
+        ASSERT_EQ(cntl._env_value, env);
+        ASSERT_EQ(cntl._partition_count, 4);
+        ASSERT_EQ(cntl._rate, 20000 / partition_count);
+        ASSERT_EQ(cntl._burstsize, 20000 / partition_count);
+        ASSERT_EQ(env_changed, true);
+        ASSERT_EQ(old_value, old_env);
+        ASSERT_EQ(parse_err, "");
+
         old_env = env;
         env = "20000*reject*100";
         ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
@@ -117,7 +129,7 @@ class token_bucket_throttling_controller_test : public ::testing::Test
         for (int i = 0; i < 100000; i++) {
             token_bucket->consumeWithBorrowAndWait(
                 1, throttle_limit / partition_count * 0.8, throttle_limit / partition_count * 1.0);
-            if (!cntl->control(1)) {
+            if (!cntl->get_token(1)) {
                 fail_count++;
             }
         }
@@ -129,7 +141,7 @@ class token_bucket_throttling_controller_test : public ::testing::Test
         for (int i = 0; i < 100000; i++) {
             token_bucket->consumeWithBorrowAndWait(
                 1, throttle_limit / partition_count * 1.2, throttle_limit / partition_count * 1.5);
-            if (!cntl->control(1)) {
+            if (!cntl->get_token(1)) {
                 fail_count++;
             }
         }
@@ -150,7 +162,7 @@ class token_bucket_throttling_controller_test : public ::testing::Test
                                                        throttle_limit / partition_count * 0.2,
                                                        throttle_limit / partition_count * 0.3);
             }
-            if (!cntl->control(1)) {
+            if (!cntl->get_token(1)) {
                 fail_count++;
             }
         }
diff --git a/src/utils/token_bucket_throttling_controller.cpp b/src/utils/token_bucket_throttling_controller.cpp
index 0d0ed2debf..8ed1ed04f3 100644
--- a/src/utils/token_bucket_throttling_controller.cpp
+++ b/src/utils/token_bucket_throttling_controller.cpp
@@ -16,6 +16,11 @@
 // under the License.
 
 #include <dsn/utils/token_bucket_throttling_controller.h>
+#include <dsn/dist/fmt_logging.h>
+#include <dsn/utility/strings.h>
+#include <dsn/utility/string_conv.h>
+#include <dsn/c/api_layer1.h>
+#include <dsn/utility/TokenBucket.h>
 
 namespace dsn {
 namespace utils {
@@ -26,15 +31,7 @@ token_bucket_throttling_controller::token_bucket_throttling_controller()
     _token_bucket = std::make_unique<DynamicTokenBucket>();
 }
 
-void token_bucket_throttling_controller::only_count(int32_t request_units)
-{
-    if (!_enabled) {
-        return;
-    }
-    _token_bucket->consumeWithBorrowNonBlocking((double)request_units, _rate, _burstsize);
-}
-
-bool token_bucket_throttling_controller::control(int32_t request_units = 1)
+bool token_bucket_throttling_controller::get_token(int32_t request_units = 1)
 {
     if (!_enabled) {
         return true;
@@ -42,10 +39,7 @@ bool token_bucket_throttling_controller::control(int32_t request_units = 1)
     auto res =
         _token_bucket->consumeWithBorrowNonBlocking((double)request_units, _rate, _burstsize);
 
-    if (res.get_value_or(0) > 0) {
-        return false;
-    }
-    return true;
+    return (res.get_value_or(0) == 0);
 }
 
 void token_bucket_throttling_controller::reset(bool &changed, std::string &old_env_value)

From bc86018e833605ceda29468387c8c0e2ea38e5cf Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Fri, 22 Oct 2021 18:09:34 +0800
Subject: [PATCH 04/14] delete header

---
 include/dsn/utils/token_bucket_throttling_controller.h | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/include/dsn/utils/token_bucket_throttling_controller.h b/include/dsn/utils/token_bucket_throttling_controller.h
index e0b1060a1c..48fa73d496 100644
--- a/include/dsn/utils/token_bucket_throttling_controller.h
+++ b/include/dsn/utils/token_bucket_throttling_controller.h
@@ -14,10 +14,15 @@
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
-
 #pragma once
 
-#include <dsn/utility/TokenBucket.h>
+#include <dsn/dist/block_service.h>
+
+namespace folly {
+template <typename Clock>
+class BasicDynamicTokenBucket;
+
+} // namespace folly
 
 namespace dsn {
 namespace utils {

From c8fd3aec025ef8d8d5ac8f7bfc854c636d2596b9 Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Fri, 22 Oct 2021 18:15:32 +0800
Subject: [PATCH 05/14] delete header

---
 include/dsn/utils/token_bucket_throttling_controller.h | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/include/dsn/utils/token_bucket_throttling_controller.h b/include/dsn/utils/token_bucket_throttling_controller.h
index 48fa73d496..fa79b6b6ba 100644
--- a/include/dsn/utils/token_bucket_throttling_controller.h
+++ b/include/dsn/utils/token_bucket_throttling_controller.h
@@ -16,7 +16,8 @@
 // under the License.
 #pragma once
 
-#include <dsn/dist/block_service.h>
+#include <chrono>
+#include <memory>
 
 namespace folly {
 template <typename Clock>

From e9eae8c3656ae7177137b52e0e66fd70c6e8aa75 Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Fri, 22 Oct 2021 18:20:48 +0800
Subject: [PATCH 06/14] change

---
 src/utils/token_bucket_throttling_controller.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/utils/token_bucket_throttling_controller.cpp b/src/utils/token_bucket_throttling_controller.cpp
index 8ed1ed04f3..d34804100c 100644
--- a/src/utils/token_bucket_throttling_controller.cpp
+++ b/src/utils/token_bucket_throttling_controller.cpp
@@ -83,7 +83,7 @@ bool token_bucket_throttling_controller::parse_from_env(const std::string &env_v
     _enabled = enabled;
     _env_value = env_value;
     _partition_count = partition_count;
-    _rate = reject_size_value / (partition_count > 0 ? partition_count : 1);
+    _rate = reject_size_value / _rate = reject_size_value / max(partition_count, 1);
     _burstsize = _rate;
     return true;
 }

From 92517eb7c5d7ad2490992af8c0b581ca99ea7d47 Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Fri, 22 Oct 2021 18:25:37 +0800
Subject: [PATCH 07/14] change

---
 src/utils/token_bucket_throttling_controller.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/utils/token_bucket_throttling_controller.cpp b/src/utils/token_bucket_throttling_controller.cpp
index d34804100c..d74aa403eb 100644
--- a/src/utils/token_bucket_throttling_controller.cpp
+++ b/src/utils/token_bucket_throttling_controller.cpp
@@ -83,7 +83,7 @@ bool token_bucket_throttling_controller::parse_from_env(const std::string &env_v
     _enabled = enabled;
     _env_value = env_value;
     _partition_count = partition_count;
-    _rate = reject_size_value / _rate = reject_size_value / max(partition_count, 1);
+    _rate = reject_size_value / max(partition_count, 1);
     _burstsize = _rate;
     return true;
 }

From f7a41296abefca49d7f620cbf80a22d59d37d63c Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Fri, 22 Oct 2021 18:40:38 +0800
Subject: [PATCH 08/14] likely

---
 src/utils/token_bucket_throttling_controller.cpp | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/src/utils/token_bucket_throttling_controller.cpp b/src/utils/token_bucket_throttling_controller.cpp
index d74aa403eb..072ef5289f 100644
--- a/src/utils/token_bucket_throttling_controller.cpp
+++ b/src/utils/token_bucket_throttling_controller.cpp
@@ -69,7 +69,8 @@ bool token_bucket_throttling_controller::parse_from_env(const std::string &env_v
     old_env_value = _env_value;
     changed = false;
 
-    if (_enabled && env_value == _env_value && partition_count == _partition_count)
+    if (_enabled && dsn_likely(env_value == _env_value) &&
+        dsn_likely(partition_count == _partition_count))
         return true;
 
     int64_t reject_size_value;

From 2ecd194a15fd81bc590281b38b5e3189c2ddbde2 Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Fri, 22 Oct 2021 18:47:51 +0800
Subject: [PATCH 09/14] update

---
 src/utils/token_bucket_throttling_controller.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/utils/token_bucket_throttling_controller.cpp b/src/utils/token_bucket_throttling_controller.cpp
index 072ef5289f..b4e6a23022 100644
--- a/src/utils/token_bucket_throttling_controller.cpp
+++ b/src/utils/token_bucket_throttling_controller.cpp
@@ -84,7 +84,7 @@ bool token_bucket_throttling_controller::parse_from_env(const std::string &env_v
     _enabled = enabled;
     _env_value = env_value;
     _partition_count = partition_count;
-    _rate = reject_size_value / max(partition_count, 1);
+    _rate = reject_size_value / std::max(partition_count, 1);
     _burstsize = _rate;
     return true;
 }

From 56964342bbc27bcc8ab0e879185fbc51dd204b5a Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Mon, 25 Oct 2021 10:32:16 +0800
Subject: [PATCH 10/14] format header

---
 src/utils/token_bucket_throttling_controller.cpp | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/src/utils/token_bucket_throttling_controller.cpp b/src/utils/token_bucket_throttling_controller.cpp
index b4e6a23022..0f5d5205c0 100644
--- a/src/utils/token_bucket_throttling_controller.cpp
+++ b/src/utils/token_bucket_throttling_controller.cpp
@@ -15,12 +15,11 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include <dsn/utils/token_bucket_throttling_controller.h>
-#include <dsn/dist/fmt_logging.h>
-#include <dsn/utility/strings.h>
-#include <dsn/utility/string_conv.h>
-#include <dsn/c/api_layer1.h>
+#include <dsn/c/api_utilities.h>
 #include <dsn/utility/TokenBucket.h>
+#include <dsn/utility/string_conv.h>
+#include <dsn/utils/token_bucket_throttling_controller.h>
+
 
 namespace dsn {
 namespace utils {

From 1559c6ec48d5d16f94a94d3e64dd7df316e21bf9 Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Mon, 25 Oct 2021 10:32:42 +0800
Subject: [PATCH 11/14] update

---
 src/utils/token_bucket_throttling_controller.cpp | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/utils/token_bucket_throttling_controller.cpp b/src/utils/token_bucket_throttling_controller.cpp
index 0f5d5205c0..003f6bc87e 100644
--- a/src/utils/token_bucket_throttling_controller.cpp
+++ b/src/utils/token_bucket_throttling_controller.cpp
@@ -20,7 +20,6 @@
 #include <dsn/utility/string_conv.h>
 #include <dsn/utils/token_bucket_throttling_controller.h>
 
-
 namespace dsn {
 namespace utils {
 

From 22c8dd0bb10b8f701c44ed20412177745b1c97f1 Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Mon, 25 Oct 2021 11:54:12 +0800
Subject: [PATCH 12/14] change

---
 include/dsn/utils/token_bucket_throttling_controller.h | 2 +-
 src/utils/token_bucket_throttling_controller.cpp       | 5 +++--
 2 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/include/dsn/utils/token_bucket_throttling_controller.h b/include/dsn/utils/token_bucket_throttling_controller.h
index fa79b6b6ba..fe2858e5f9 100644
--- a/include/dsn/utils/token_bucket_throttling_controller.h
+++ b/include/dsn/utils/token_bucket_throttling_controller.h
@@ -79,7 +79,7 @@ class token_bucket_throttling_controller
     // 'changed' is set when return true.
     // 'old_env_value' is set when 'changed' is set to true.
     bool parse_from_env(const std::string &env_value,
-                        int partition_count,
+                        int32_t partition_count,
                         std::string &parse_error,
                         bool &changed,
                         std::string &old_env_value);
diff --git a/src/utils/token_bucket_throttling_controller.cpp b/src/utils/token_bucket_throttling_controller.cpp
index 003f6bc87e..4d8714ff1b 100644
--- a/src/utils/token_bucket_throttling_controller.cpp
+++ b/src/utils/token_bucket_throttling_controller.cpp
@@ -59,7 +59,7 @@ void token_bucket_throttling_controller::reset(bool &changed, std::string &old_e
 const std::string &token_bucket_throttling_controller::env_value() const { return _env_value; }
 
 bool token_bucket_throttling_controller::parse_from_env(const std::string &env_value,
-                                                        int partition_count,
+                                                        int32_t partition_count,
                                                         std::string &parse_error,
                                                         bool &changed,
                                                         std::string &old_env_value)
@@ -68,8 +68,9 @@ bool token_bucket_throttling_controller::parse_from_env(const std::string &env_v
     changed = false;
 
     if (_enabled && dsn_likely(env_value == _env_value) &&
-        dsn_likely(partition_count == _partition_count))
+        dsn_likely(partition_count == _partition_count)) {
         return true;
+    }
 
     int64_t reject_size_value;
     bool enabled;

From b3e8141dc5b7c5583535e1bfa985e40b90d4119a Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Tue, 26 Oct 2021 12:14:55 +0800
Subject: [PATCH 13/14] add some unit tests

---
 .../token_bucket_throttling_controller.h      |   4 +-
 ...oken_bucket_throttling_controller_test.cpp | 141 ++++++++----------
 2 files changed, 64 insertions(+), 81 deletions(-)

diff --git a/include/dsn/utils/token_bucket_throttling_controller.h b/include/dsn/utils/token_bucket_throttling_controller.h
index fe2858e5f9..363a4d72b3 100644
--- a/include/dsn/utils/token_bucket_throttling_controller.h
+++ b/include/dsn/utils/token_bucket_throttling_controller.h
@@ -65,12 +65,12 @@ class token_bucket_throttling_controller
     //      example: 20000*delay*100,20000*reject*100
     //      result: reject 20000 request_units, but never delay
     //      example: 20000*delay*100
-    //      result: never result or delay
+    //      result: never reject or delay
     //      example: 20000*reject*100
     //      result: reject 20000 request_units
     // 2. style: 20/"20K"/"20M"
     //      example: 20K
-    //      result reject 20000 request_units
+    //      result: reject 20000 request_units
     //
     // return true if parse succeed.
     // return false if parse failed for the reason of invalid env_value.
diff --git a/src/utils/test/token_bucket_throttling_controller_test.cpp b/src/utils/test/token_bucket_throttling_controller_test.cpp
index 07d325300d..dff69ae7b3 100644
--- a/src/utils/test/token_bucket_throttling_controller_test.cpp
+++ b/src/utils/test/token_bucket_throttling_controller_test.cpp
@@ -23,93 +23,76 @@
 namespace dsn {
 namespace utils {
 
+#define INVALIDATE_SITUATION_CHECK(env)                                                            \
+    do {                                                                                           \
+        std::string old_value, parse_err;                                                          \
+        bool env_changed_result = false;                                                           \
+        ASSERT_FALSE(cntl.parse_from_env(env, 4, parse_err, env_changed_result, old_value));       \
+        ASSERT_EQ(env_changed_result, false);                                                      \
+        ASSERT_EQ(parse_err, "wrong format, you can set like 20000 or 20K");                       \
+        ASSERT_EQ(cntl._enabled, true);                                                            \
+        ASSERT_EQ(old_value, old_env);                                                             \
+        ASSERT_EQ(cntl._env_value, old_env);                                                       \
+    } while (0)
+
+#define VALIDATE_SITUATION_CHECK(                                                                  \
+    env, partition_count, throttle_size, enabled, env_changed, old_env)                            \
+    do {                                                                                           \
+        bool env_changed_result = false;                                                           \
+        std::string old_value, parse_err;                                                          \
+        int32_t partitioned_throttle_size = throttle_size / partition_count;                       \
+        ASSERT_TRUE(                                                                               \
+            cntl.parse_from_env(env, partition_count, parse_err, env_changed_result, old_value));  \
+        ASSERT_EQ(cntl._env_value, env);                                                           \
+        ASSERT_EQ(cntl._partition_count, partition_count);                                         \
+        ASSERT_EQ(cntl._burstsize, partitioned_throttle_size);                                     \
+        ASSERT_EQ(cntl._rate, partitioned_throttle_size);                                          \
+        ASSERT_EQ(cntl._enabled, enabled);                                                         \
+        ASSERT_EQ(env_changed_result, env_changed);                                                \
+        ASSERT_EQ(old_value, old_env);                                                             \
+        ASSERT_EQ(parse_err, "");                                                                  \
+    } while (0)
+
 class token_bucket_throttling_controller_test : public ::testing::Test
 {
 public:
     void test_parse_env_basic_token_bucket_throttling()
     {
         token_bucket_throttling_controller cntl;
-        std::string parse_err;
-        bool env_changed = false;
-        std::string old_value;
 
-        int partition_count = 4;
-        std::string old_env = "";
-        std::string env = "20000*delay*100";
         // token_bucket_throttling_controller doesn't support delay only
-        ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
-        ASSERT_EQ(cntl._env_value, env);
-        ASSERT_EQ(cntl._partition_count, 4);
-        ASSERT_EQ(cntl._burstsize, 0);
-        ASSERT_EQ(cntl._rate, 0);
-        ASSERT_EQ(cntl._enabled, false);
-        ASSERT_EQ(env_changed, true);
-        ASSERT_EQ(old_value, old_env);
-        ASSERT_EQ(parse_err, "");
-
-        old_env = env;
-        env = "200K";
-        ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
-        ASSERT_EQ(cntl._enabled, true);
-        ASSERT_EQ(cntl._env_value, env);
-        ASSERT_EQ(cntl._partition_count, 4);
-        ASSERT_EQ(cntl._rate, 200000 / partition_count);
-        ASSERT_EQ(cntl._burstsize, 200000 / partition_count);
-        ASSERT_EQ(env_changed, true);
-        ASSERT_EQ(old_value, old_env);
-        ASSERT_EQ(parse_err, "");
-
-        old_env = env;
-        env = "20000*delay*100,20000*reject*100";
-        ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
-        ASSERT_EQ(cntl._enabled, true);
-        ASSERT_EQ(cntl._env_value, env);
-        ASSERT_EQ(cntl._partition_count, 4);
-        ASSERT_EQ(cntl._rate, 20000 / partition_count);
-        ASSERT_EQ(cntl._burstsize, 20000 / partition_count);
-        ASSERT_EQ(env_changed, true);
-        ASSERT_EQ(old_value, old_env);
-        ASSERT_EQ(parse_err, "");
-
-        old_env = env;
-        env = "20K*delay*100,20K*reject*100";
-        ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
-        ASSERT_EQ(cntl._enabled, true);
-        ASSERT_EQ(cntl._env_value, env);
-        ASSERT_EQ(cntl._partition_count, 4);
-        ASSERT_EQ(cntl._rate, 20000 / partition_count);
-        ASSERT_EQ(cntl._burstsize, 20000 / partition_count);
-        ASSERT_EQ(env_changed, true);
-        ASSERT_EQ(old_value, old_env);
-        ASSERT_EQ(parse_err, "");
-
-        old_env = env;
-        env = "20000*reject*100";
-        ASSERT_TRUE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
-        ASSERT_EQ(cntl._enabled, true);
-        ASSERT_EQ(cntl._env_value, env);
-        ASSERT_EQ(cntl._partition_count, 4);
-        ASSERT_EQ(cntl._rate, 20000 / partition_count);
-        ASSERT_EQ(cntl._burstsize, 20000 / partition_count);
-        ASSERT_EQ(env_changed, true);
-        ASSERT_EQ(old_value, old_env);
-        ASSERT_EQ(parse_err, "");
-
-        // invalid argument
-        old_env = env;
-        env = "*deldday*100";
-        ASSERT_FALSE(cntl.parse_from_env(env, partition_count, parse_err, env_changed, old_value));
-        ASSERT_EQ(env_changed, false);
-        ASSERT_EQ(parse_err, "wrong format, you can set like 20000 or 20K");
-        ASSERT_EQ(cntl._enabled, true); // ensure invalid env won't stop throttling
-        ASSERT_EQ(old_value, old_env);
-
-        ASSERT_FALSE(cntl.parse_from_env("", 4, parse_err, env_changed, old_value));
-        ASSERT_EQ(env_changed, false);
-        ASSERT_NE(parse_err, "");
-        ASSERT_EQ(parse_err, "wrong format, you can set like 20000 or 20K");
-        ASSERT_EQ(cntl._enabled, true);
-        ASSERT_EQ(old_value, old_env);
+        VALIDATE_SITUATION_CHECK("20000*delay*100", 4, 0, false, true, "");
+        VALIDATE_SITUATION_CHECK("200K", 4, 200000, true, true, "20000*delay*100");
+        VALIDATE_SITUATION_CHECK("20000*delay*100,20000*reject*100", 4, 20000, true, true, "200K");
+        VALIDATE_SITUATION_CHECK("20K*delay*100,20K*reject*100",
+                                 4,
+                                 20000,
+                                 true,
+                                 true,
+                                 "20000*delay*100,20000*reject*100");
+        VALIDATE_SITUATION_CHECK(
+            "20000*reject*100", 4, 20000, true, true, "20K*delay*100,20K*reject*100");
+
+
+        // invalid argument]
+        std::string old_env = "20000*reject*100";
+        INVALIDATE_SITUATION_CHECK("*deldday*100");
+        INVALIDATE_SITUATION_CHECK("");
+        INVALIDATE_SITUATION_CHECK("*reject");
+        INVALIDATE_SITUATION_CHECK("*reject*");
+        INVALIDATE_SITUATION_CHECK("reject*");
+        INVALIDATE_SITUATION_CHECK("reject");
+        INVALIDATE_SITUATION_CHECK("200g");
+        INVALIDATE_SITUATION_CHECK("200G");
+        INVALIDATE_SITUATION_CHECK("M");
+        INVALIDATE_SITUATION_CHECK("K");
+        INVALIDATE_SITUATION_CHECK("-1K");
+        INVALIDATE_SITUATION_CHECK("1aK");
+        INVALIDATE_SITUATION_CHECK("pegNo1");
+        INVALIDATE_SITUATION_CHECK("-20");
+        INVALIDATE_SITUATION_CHECK("12KM");
+        INVALIDATE_SITUATION_CHECK("1K2M");
+        INVALIDATE_SITUATION_CHECK("2000K0*reject*100");
     }
 
     void throttle_test()

From 46d4337824cc558713455f81cd5d2da8b28b1628 Mon Sep 17 00:00:00 2001
From: tang yanzhao <smityz@qq.com>
Date: Tue, 26 Oct 2021 12:20:53 +0800
Subject: [PATCH 14/14] format

---
 src/utils/test/token_bucket_throttling_controller_test.cpp | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/utils/test/token_bucket_throttling_controller_test.cpp b/src/utils/test/token_bucket_throttling_controller_test.cpp
index dff69ae7b3..1b20715225 100644
--- a/src/utils/test/token_bucket_throttling_controller_test.cpp
+++ b/src/utils/test/token_bucket_throttling_controller_test.cpp
@@ -73,7 +73,6 @@ class token_bucket_throttling_controller_test : public ::testing::Test
         VALIDATE_SITUATION_CHECK(
             "20000*reject*100", 4, 20000, true, true, "20K*delay*100,20K*reject*100");
 
-
         // invalid argument]
         std::string old_env = "20000*reject*100";
         INVALIDATE_SITUATION_CHECK("*deldday*100");