This is an automated email from the ASF dual-hosted git repository. alexey pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/kudu.git
The following commit(s) were added to refs/heads/master by this push: new 274eadfd7 [util] a small clean up on the Throttler class 274eadfd7 is described below commit 274eadfd79cb2ffcaa0c016376469d70036e11ca Author: Alexey Serbin <ale...@apache.org> AuthorDate: Fri Jul 19 15:38:57 2024 -0700 [util] a small clean up on the Throttler class The motivation behind this patch was an idea of making the API of the Throttler class more robust after reviewing a changelist that used the functionality of the Throttler. I have doubts that the current implementation of the Throttler behaves the way one would expect (there is just spotty test coverage for that), but it's another story. Change-Id: I43d60323c3d84da896c1a5429dfb7d461a24f9b2 Reviewed-on: http://gerrit.cloudera.org:8080/21603 Reviewed-by: Yingchun Lai <laiyingc...@apache.org> Tested-by: Alexey Serbin <ale...@apache.org> --- src/kudu/tablet/tablet.cc | 5 ++- src/kudu/tools/table_scanner.cc | 6 ++-- src/kudu/tools/tool_action_local_replica.cc | 3 +- src/kudu/tserver/tablet_copy_client-test.cc | 1 - src/kudu/tserver/tablet_copy_client.cc | 4 +-- src/kudu/util/throttler-test.cc | 22 ++++++++---- src/kudu/util/throttler.cc | 48 ++++++++++++++++--------- src/kudu/util/throttler.h | 55 ++++++++++++++++++----------- 8 files changed, 89 insertions(+), 55 deletions(-) diff --git a/src/kudu/tablet/tablet.cc b/src/kudu/tablet/tablet.cc index d6d811d6a..909d5df0e 100644 --- a/src/kudu/tablet/tablet.cc +++ b/src/kudu/tablet/tablet.cc @@ -386,8 +386,7 @@ Tablet::Tablet(scoped_refptr<TabletMetadata> metadata, FLAGS_tablet_compaction_budget_mb, metrics_.get())); if (FLAGS_tablet_throttler_rpc_per_sec > 0 || FLAGS_tablet_throttler_bytes_per_sec > 0) { - throttler_.reset(new Throttler(MonoTime::Now(), - FLAGS_tablet_throttler_rpc_per_sec, + throttler_.reset(new Throttler(FLAGS_tablet_throttler_rpc_per_sec, FLAGS_tablet_throttler_bytes_per_sec, FLAGS_tablet_throttler_burst_factor)); } @@ -1795,7 +1794,7 @@ bool Tablet::ShouldThrottleAllow(int64_t bytes) { if (!throttler_) { return true; } - return throttler_->Take(MonoTime::Now(), 1, bytes); + return throttler_->Take(1, bytes); } Status Tablet::PickRowSetsToCompact(RowSetsInCompactionOrFlush *picked, diff --git a/src/kudu/tools/table_scanner.cc b/src/kudu/tools/table_scanner.cc index eebb94a41..fac0ce949 100644 --- a/src/kudu/tools/table_scanner.cc +++ b/src/kudu/tools/table_scanner.cc @@ -582,7 +582,7 @@ TableScanner::TableScanner( out_(nullptr) { CHECK_OK(SetReplicaSelection(FLAGS_replica_selection)); if (FLAGS_table_copy_throttler_bytes_per_sec > 0) { - throttler_ = std::make_shared<Throttler>(MonoTime::Now(), 0, + throttler_ = std::make_shared<Throttler>(Throttler::kNoLimit, FLAGS_table_copy_throttler_bytes_per_sec, FLAGS_table_copy_throttler_burst_factor); } @@ -611,9 +611,9 @@ Status TableScanner::ScanData(const vector<KuduScanToken*>& tokens, // Limit table copy speed. if (throttler_) { SCOPED_LOG_SLOW_EXECUTION(WARNING, 1000, "Table copy throttler"); - while (!throttler_->Take(MonoTime::Now(), 0, + while (!throttler_->Take(0, batch.direct_data().size() + batch.indirect_data().size())) { - SleepFor(MonoDelta::FromMilliseconds(10)); + SleepFor(MonoDelta::FromMicroseconds(Throttler::kRefillPeriodMicros / 2)); } } RETURN_NOT_OK(cb(batch)); diff --git a/src/kudu/tools/tool_action_local_replica.cc b/src/kudu/tools/tool_action_local_replica.cc index 2449db455..5c15d6bbd 100644 --- a/src/kudu/tools/tool_action_local_replica.cc +++ b/src/kudu/tools/tool_action_local_replica.cc @@ -334,8 +334,7 @@ class TabletCopier { shared_ptr<Throttler> throttler; if (FLAGS_tablet_copy_throttler_bytes_per_sec > 0) { - throttler = std::make_shared<Throttler>(MonoTime::Now(), - 0, + throttler = std::make_shared<Throttler>(0, FLAGS_tablet_copy_throttler_bytes_per_sec, FLAGS_tablet_copy_throttler_burst_factor); } diff --git a/src/kudu/tserver/tablet_copy_client-test.cc b/src/kudu/tserver/tablet_copy_client-test.cc index a53f2aa31..fadaca536 100644 --- a/src/kudu/tserver/tablet_copy_client-test.cc +++ b/src/kudu/tserver/tablet_copy_client-test.cc @@ -319,7 +319,6 @@ class TabletCopyThrottlerTest : public TabletCopyClientTest { TabletCopyThrottlerTest() { mode_ = TabletCopyMode::REMOTE; throttler_ = std::make_shared<Throttler>( - MonoTime::Now(), 0, FLAGS_tablet_copy_transfer_chunk_size_bytes, 2 * FLAGS_tablet_copy_transfer_chunk_size_bytes); diff --git a/src/kudu/tserver/tablet_copy_client.cc b/src/kudu/tserver/tablet_copy_client.cc index d392c72c2..85772f813 100644 --- a/src/kudu/tserver/tablet_copy_client.cc +++ b/src/kudu/tserver/tablet_copy_client.cc @@ -1036,8 +1036,8 @@ Status RemoteTabletCopyClient::DownloadFile(const DataIdPB& data_id, } if (throttler_) { LOG_TIMING(INFO, "Tablet copy throttler") { - while (!throttler_->Take(MonoTime::Now(), 0, chunk_size)) { - SleepFor(MonoDelta::FromMilliseconds(10)); + while (!throttler_->Take(0, chunk_size)) { + SleepFor(MonoDelta::FromMicroseconds(Throttler::kRefillPeriodMicros / 2)); } } } diff --git a/src/kudu/util/throttler-test.cc b/src/kudu/util/throttler-test.cc index ff97eb5da..3c391949b 100644 --- a/src/kudu/util/throttler-test.cc +++ b/src/kudu/util/throttler-test.cc @@ -17,6 +17,8 @@ #include "kudu/util/throttler.h" +#include <string> + #include <gtest/gtest.h> #include "kudu/util/monotime.h" @@ -27,7 +29,13 @@ namespace kudu { class ThrottlerTest : public KuduTest { }; -TEST_F(ThrottlerTest, TestOpThrottle) { +TEST_F(ThrottlerTest, Basic) { + Throttler t(1, 1, 1.0); + ASSERT_TRUE(t.Take(0, 1)); + ASSERT_TRUE(t.Take(1, 0)); +} + +TEST_F(ThrottlerTest, OpThrottle) { // Check operation rate throttling MonoTime now = MonoTime::Now(); Throttler t0(now, 1000, 1000*1000, 1); @@ -39,12 +47,12 @@ TEST_F(ThrottlerTest, TestOpThrottle) { ASSERT_TRUE(t0.Take(now, 1, 1)); } ASSERT_FALSE(t0.Take(now, 1, 1)); - now += MonoDelta::FromMilliseconds(100); + now += MonoDelta::FromMicroseconds(Throttler::kRefillPeriodMicros); } } -TEST_F(ThrottlerTest, TestIOThrottle) { - // Check operation rate throttling +TEST_F(ThrottlerTest, IOThrottle) { + // Check IO rate throttling MonoTime now = MonoTime::Now(); Throttler t0(now, 50000, 1000*1000, 1); // Fill up bucket @@ -55,12 +63,12 @@ TEST_F(ThrottlerTest, TestIOThrottle) { ASSERT_TRUE(t0.Take(now, 1, 1000)); } ASSERT_FALSE(t0.Take(now, 1, 1000)); - now += MonoDelta::FromMilliseconds(100); + now += MonoDelta::FromMilliseconds(Throttler::kRefillPeriodMicros); } } -TEST_F(ThrottlerTest, TestBurst) { - // Check IO rate throttling +TEST_F(ThrottlerTest, Burst) { + // Check throttling for bursty consuming MonoTime now = MonoTime::Now(); Throttler t0(now, 2000, 1000*1000, 5); // Fill up bucket diff --git a/src/kudu/util/throttler.cc b/src/kudu/util/throttler.cc index 69e0f99dc..f4b35428c 100644 --- a/src/kudu/util/throttler.cc +++ b/src/kudu/util/throttler.cc @@ -18,33 +18,49 @@ #include "kudu/util/throttler.h" #include <algorithm> -#include <mutex> + +#include <glog/logging.h> namespace kudu { -Throttler::Throttler(MonoTime now, uint64_t op_rate, uint64_t byte_rate, double burst_factor) : - next_refill_(now) { - op_refill_ = op_rate / (MonoTime::kMicrosecondsPerSecond / kRefillPeriodMicros); - op_token_ = 0; - op_token_max_ = static_cast<uint64_t>(op_refill_ * burst_factor); - byte_refill_ = byte_rate / (MonoTime::kMicrosecondsPerSecond / kRefillPeriodMicros); - byte_token_ = 0; - byte_token_max_ = static_cast<uint64_t>(byte_refill_ * burst_factor); +Throttler::Throttler(uint64_t op_rate_per_sec, + uint64_t byte_rate_per_sec, + double burst_factor) + : Throttler(MonoTime::Now(), op_rate_per_sec, byte_rate_per_sec, burst_factor) { +} + +Throttler::Throttler(MonoTime now, + uint64_t op_rate_per_sec, + uint64_t byte_rate_per_sec, + double burst_factor) + : byte_refill_(byte_rate_per_sec / (MonoTime::kMicrosecondsPerSecond / kRefillPeriodMicros)), + byte_token_max_(static_cast<uint64_t>(byte_refill_ * burst_factor)), + op_refill_(op_rate_per_sec / (MonoTime::kMicrosecondsPerSecond / kRefillPeriodMicros)), + op_token_max_(static_cast<uint64_t>(op_refill_ * burst_factor)), + byte_token_(0), + op_token_(0), + next_refill_(now) { } -bool Throttler::Take(MonoTime now, uint64_t op, uint64_t byte) { - if (op_refill_ == 0 && byte_refill_ == 0) { +bool Throttler::Take(uint64_t ops, uint64_t bytes) { + return Take(MonoTime::Now(), ops, bytes); +} + +bool Throttler::Take(MonoTime now, uint64_t ops, uint64_t bytes) { + DCHECK(ops > 0 || bytes > 0); + if (op_refill_ == kNoLimit && byte_refill_ == kNoLimit) { return true; } - std::lock_guard<simple_spinlock> lock(lock_); + + std::lock_guard lock(lock_); Refill(now); - if ((op_refill_ == 0 || op <= op_token_) && - (byte_refill_ == 0 || byte <= byte_token_)) { + if ((op_refill_ == 0 || ops <= op_token_) && + (byte_refill_ == 0 || bytes <= byte_token_)) { if (op_refill_ > 0) { - op_token_ -= op; + op_token_ -= ops; } if (byte_refill_ > 0) { - byte_token_ -= byte; + byte_token_ -= bytes; } return true; } diff --git a/src/kudu/util/throttler.h b/src/kudu/util/throttler.h index 559409120..191408077 100644 --- a/src/kudu/util/throttler.h +++ b/src/kudu/util/throttler.h @@ -14,49 +14,62 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. -#ifndef KUDU_UTIL_THROTTLER_H -#define KUDU_UTIL_THROTTLER_H +#pragma once #include <cstdint> +#include <gtest/gtest_prod.h> + #include "kudu/util/locks.h" #include "kudu/util/monotime.h" namespace kudu { // A throttler to throttle both operation/s and IO byte/s. -class Throttler { +class Throttler final { public: // Refill period is 100ms. - enum { - kRefillPeriodMicros = 100000 - }; + static constexpr const int64_t kRefillPeriodMicros = 100000; + static constexpr const uint64_t kNoLimit = 0; // Construct a throttler with max operation per second, max IO bytes per second - // and burst factor (burst_rate = rate * burst), burst rate means maximum - // throughput within one refill period. - // Set op_per_sec to 0 to disable operation throttling. - // Set byte_per_sec to 0 to disable IO bytes throttling. - Throttler(MonoTime now, uint64_t op_per_sec, uint64_t byte_per_sec, double burst_factor); + // and burst factor (burst_rate = rate * burst_factor), burst_rate means + // the maximum throughput within one refill period. + // Set op_rate_per_sec to kNoLimit to disable operation rate throttling. + // Set byte_rate_per_sec to kNoLimit to disable IO rate throttling. + Throttler(uint64_t op_rate_per_sec, + uint64_t byte_rate_per_sec, + double burst_factor); - // Throttle an "operation group" by taking 'op' operation tokens and 'byte' byte tokens. + // Throttle an "operation group" by taking 'ops' operation tokens and 'bytes' + // byte tokens. // Return true if there are enough tokens, and operation is allowed. // Return false if there are not enough tokens, and operation is throttled. - bool Take(MonoTime now, uint64_t op, uint64_t byte); + bool Take(uint64_t ops, uint64_t bytes); private: + FRIEND_TEST(ThrottlerTest, OpThrottle); + FRIEND_TEST(ThrottlerTest, IOThrottle); + FRIEND_TEST(ThrottlerTest, Burst); + + Throttler(MonoTime now, + uint64_t op_rate_per_sec, + uint64_t byte_rate_per_sec, + double burst_factor); + + bool Take(MonoTime now, uint64_t ops, uint64_t bytes); + void Refill(MonoTime now); - MonoTime next_refill_; - uint64_t op_refill_; - uint64_t op_token_; - uint64_t op_token_max_; - uint64_t byte_refill_; + const uint64_t byte_refill_; + const uint64_t byte_token_max_; + const uint64_t op_refill_; + const uint64_t op_token_max_; + uint64_t byte_token_; - uint64_t byte_token_max_; + uint64_t op_token_; + MonoTime next_refill_; simple_spinlock lock_; }; } // namespace kudu - -#endif